@Bean
public ProducerFactory<String, Object> highThroughputProducerFactory() {
    Map<String, Object> config = new HashMap<>();
 
    // THROUGHPUT settings
    config.put(ProducerConfig.LINGER_MS_CONFIG, 20);
    // Wait up to 20ms to accumulate messages before sending
    // Allows batching at the cost of slightly higher latency
    // Default is 0 (send immediately) - terrible for throughput
 
    config.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);
    // 64KB batch size (default is 16KB)
    // Larger batches = more messages per network round trip
 
    config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864L);
    // 64MB total buffer memory (default is 32MB)
    // Producer blocks when buffer is full
 
    config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
    // Compress batches before sending
    // LZ4 is fast with good compression ratio
 
    // RELIABILITY settings (do not sacrifice these for throughput)
    config.put(ProducerConfig.ACKS_CONFIG, "1");
    // acks=1 for throughput (leader ACK only)
    // Use acks=all for critical data
 
    config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    // 5 in-flight requests per connection
    // Increases throughput but can affect ordering on retry
    // Set to 1 if you need strict per-key ordering with retries
 
    return new DefaultKafkaProducerFactory<>(config);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> factory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
 
    factory.setConcurrency(6);
    // 6 concurrent consumer threads per instance
    // Each thread processes one partition
    // Requires at least 6 partitions for full utilization
 
    return factory;
}

@KafkaListener(topics = "image-resize-jobs", concurrency = "6")
public void processImage(ImageResizeJob job, Acknowledgment ack) {
    CompletableFuture.runAsync(() -> {
        imageService.resize(job.getImageUrl(), job.getTargetDimensions());
    }, imageProcessingExecutor)
    .thenRun(ack::acknowledge)
    .exceptionally(e -> {
        log.error("Image resize failed: {}", job.getImageId(), e);
        return null; // Do not ACK - will be redelivered
    });
}

// Slow - N database round trips for N messages
@KafkaListener(topics = "user-events")
public void handleUserEvent(UserEvent event) {
    userRepository.findById(event.getUserId())  // Database call
        .ifPresent(user -> {
            userRepository.save(user.apply(event));  // Another database call
        });
}
 
// Fast - 1 database round trip for N messages (batch consumer)
@KafkaListener(topics = "user-events", batch = "true")
public void handleUserEvents(List<UserEvent> events) {
    Set<String> userIds = events.stream()
        .map(UserEvent::getUserId)
        .collect(Collectors.toSet());
 
    // Single query for all users in the batch
    Map<String, User> users = userRepository.findAllById(userIds)
        .stream().collect(Collectors.toMap(User::getId, u -> u));
 
    List<User> updatedUsers = events.stream()
        .map(event -> users.get(event.getUserId()).apply(event))
        .collect(Collectors.toList());
 
    // Single batch save
    userRepository.saveAll(updatedUsers);
}

// Single message processing - 1000 DB calls for 1000 messages
@KafkaListener(topics = "orders")
public void process(OrderEvent event) {
    orderRepository.save(new ProcessedOrder(event));  // 1 DB call per message
}
 
// Batch processing - 10 DB calls for 1000 messages (100x reduction)
@KafkaListener(topics = "orders", batch = "true")
public void processBatch(List<ConsumerRecord<String, OrderEvent>> records) {
    List<ProcessedOrder> orders = records.stream()
        .map(record -> new ProcessedOrder(record.value()))
        .collect(Collectors.toList());
 
    orderRepository.saveAll(orders);  // 1 DB call for the whole batch
 
    // Acknowledge all records in the batch
    // Spring Kafka handles this automatically in batch mode
}

# Maximum records to fetch per poll
spring.kafka.consumer.max-poll-records=500
 
# Maximum bytes to fetch per poll per partition
fetch.max.bytes=52428800     # 50MB
fetch.min.bytes=1024         # Wait for at least 1KB before returning
fetch.max.wait.ms=500        # Wait at most 500ms for fetch.min.bytes

// Manual batching in RabbitMQ
@Service
public class BatchPublisher {
 
    private final List<Message> pendingBatch = new ArrayList<>();
    private static final int BATCH_SIZE = 100;
 
    public synchronized void addToBatch(Object payload) {
        pendingBatch.add(MessageBuilder.withBody(
            objectMapper.writeValueAsBytes(payload)).build());
 
        if (pendingBatch.size() >= BATCH_SIZE) {
            flush();
        }
    }
 
    @Scheduled(fixedDelay = 1000)
    public synchronized void flush() {
        if (pendingBatch.isEmpty()) return;
 
        rabbitTemplate.execute(channel -> {
            for (Message message : pendingBatch) {
                channel.basicPublish("exchange", "routing.key", null, message.getBody());
            }
            channel.waitForConfirmsOrDie(5000);  // Wait for broker ACK
            return null;
        });
 
        pendingBatch.clear();
    }
}

// Kafka compression (batch-level)
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
// Entire batch is compressed as one unit
// More messages in batch = better compression ratio

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> factory() {
    factory.getContainerProperties().setPartitionAssignmentStrategy(
        List.of(CooperativeStickyAssignor.class)
    );
    // CooperativeStickyAssignor (Kafka 2.4+):
    // - Assigns only the necessary partitions, not all
    // - Consumers keep their current partitions when possible
    // - No stop-the-world rebalance - incremental assignment
    // - Greatly reduces impact of scaling events
}
 
// Tune session timeout and heartbeat
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);      // 45 seconds
config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000);   // 15 seconds
config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);   // 5 minutes max between polls

# Kafka CLI
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --group email-service-group \
  --describe
 
# Output:
# TOPIC         PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# order-events  0          49900           50000           100  email-consumer-1
# order-events  1          50200           50200           0    email-consumer-2
# order-events  2          49800           50000           200  email-consumer-3
# TOTAL LAG: 300 messages

// Programmatic lag monitoring
@Service
public class ConsumerLagMonitor {
 
    private final AdminClient adminClient;
    private final MeterRegistry meterRegistry;
 
    @Scheduled(fixedDelay = 30000)
    public void reportLag() {
        Map<TopicPartition, OffsetAndMetadata> offsets = adminClient
            .listConsumerGroupOffsets("email-service-group")
            .partitionsToOffsetAndMetadata()
            .get();
 
        Map<TopicPartition, Long> endOffsets = adminClient
            .listOffsets(offsets.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())))
            .all()
            .get()
            .entrySet().stream()
            .collect(Collectors.toMap(Map.Entry::getKey,
                e -> e.getValue().offset()));
 
        long totalLag = offsets.entrySet().stream()
            .mapToLong(entry -> {
                long endOffset = endOffsets.getOrDefault(entry.getKey(), 0L);
                long consumerOffset = entry.getValue().offset();
                long lag = Math.max(0, endOffset - consumerOffset);
 
                // Report per-partition lag as a gauge metric
                meterRegistry.gauge("kafka.consumer.lag",
                    Tags.of("topic", entry.getKey().topic(),
                            "partition", String.valueOf(entry.getKey().partition()),
                            "group", "email-service-group"),
                    lag);
 
                return lag;
            }).sum();
 
        // Alert if total lag exceeds threshold
        if (totalLag > 10000) {
            alertingService.sendAlert("Consumer lag exceeds 10,000: current lag = " + totalLag);
        }
    }
}

# Producer Metrics
kafka.producer.record-send-rate:          Messages sent per second
kafka.producer.record-error-rate:         Failed sends per second
kafka.producer.request-latency-avg:       Average time for broker to ACK
kafka.producer.batch-size-avg:            Average batch size (bigger = more efficient)
kafka.producer.compression-rate:          Compression effectiveness (lower = better)
kafka.producer.record-queue-time-avg:     Time messages wait in buffer before sending
 
# Consumer Metrics
kafka.consumer.records-consumed-rate:     Messages consumed per second
kafka.consumer.fetch-rate:                Poll calls per second
kafka.consumer.fetch-latency-avg:         Time to fetch from broker
kafka.consumer.records-lag:               MOST IMPORTANT: messages behind head
kafka.consumer.records-lag-max:           Maximum lag across all partitions
kafka.consumer.commit-latency-avg:        Time to commit offsets
 
# Broker Metrics
kafka.server.bytes-in-per-sec:            Incoming data rate
kafka.server.bytes-out-per-sec:           Outgoing data rate
kafka.server.messages-in-per-sec:         Message ingestion rate
kafka.server.request-handler-pool.idle:   Broker CPU availability
kafka.log.flush-time-ms-p99:              Disk flush latency (99th percentile)
kafka.replica.under-replicated-partitions: > 0 means replication is lagging (CRITICAL)
kafka.controller.active-controller-count: Must be exactly 1

rabbitmq.queue.messages:               Total messages in queue
rabbitmq.queue.messages.ready:         Messages waiting for consumer
rabbitmq.queue.messages.unacknowledged: Messages in flight (consumer has but not ACKed)
rabbitmq.queue.consumers:              Number of active consumers
rabbitmq.queue.message.publish.rate:   Messages published per second
rabbitmq.queue.message.deliver.rate:   Messages delivered per second
rabbitmq.queue.message.ack.rate:       Messages acknowledged per second
rabbitmq.connection.count:             Active connections
rabbitmq.channel.count:                Active channels
rabbitmq.node.disk.free.alarm:         CRITICAL: disk space alarm
rabbitmq.node.mem.alarm:               CRITICAL: memory alarm

# Assign brokers to different racks/availability zones
# Kafka distributes replicas across racks for AZ resilience
broker.rack=us-east-1a  # Broker 1 configuration
broker.rack=us-east-1b  # Broker 2 configuration
broker.rack=us-east-1c  # Broker 3 configuration
 
# Result: even if AZ us-east-1a goes down,
# all partitions still have replicas in 1b and 1c

// Quorum Queues - Raft-based, strongly consistent replication
@Bean
public Queue highAvailabilityQueue() {
    return QueueBuilder.durable("critical-orders")
        .withArgument("x-queue-type", "quorum")
        // quorum queues:
        // - Automatically replicated across cluster nodes
        // - Data not lost if minority of nodes fail
        // - Default replication: majority of nodes
        // - No configuration of replica count needed (uses cluster size)
        .build();
}

# Per-topic retention - set when creating topic
kafka-topics.sh --create \
  --topic order-events \
  --config retention.ms=604800000 \    # 7 days by default
  --config retention.bytes=10737418240 \ # 10GB max per partition
  --config segment.ms=86400000 \       # Roll log segment daily
  --config cleanup.policy=delete        # delete (default) or compact
 
# Modify retention on existing topic
kafka-configs.sh \
  --bootstrap-server kafka:9092 \
  --entity-type topics \
  --entity-name order-events \
  --alter \
  --add-config retention.ms=2592000000  # Change to 30 days

// Per-message TTL
MessageProperties props = new MessageProperties();
props.setExpiration("60000");  // Milliseconds - expires after 60 seconds
Message message = new Message(payload, props);
rabbitTemplate.send("exchange", "routing.key", message);
 
// Per-queue TTL - all messages in this queue expire after 5 minutes
@Bean
public Queue timeSensitiveQueue() {
    return QueueBuilder.durable("time-sensitive-notifications")
        .withArgument("x-message-ttl", 300000)   // 5 minutes in milliseconds
        .withArgument("x-dead-letter-exchange", "dlx.expired")  // Send expired to DLQ
        .build();
}

// Maximum retention: 14 days
CreateQueueRequest request = CreateQueueRequest.builder()
    .queueName("order-processing")
    .attributes(Map.of(
        QueueAttributeName.MESSAGE_RETENTION_PERIOD, "1209600",  // 14 days in seconds
        QueueAttributeName.VISIBILITY_TIMEOUT, "300",            // 5 minute visibility timeout
        QueueAttributeName.RECEIVE_MESSAGE_WAIT_TIME_SECONDS, "20" // Long polling
    ))
    .build();

# Producer - High Throughput Mode
batch.size=131072           # 128KB - larger batches
linger.ms=50                # 50ms batching window
compression.type=lz4        # Fast compression
acks=1                      # Leader ACK only (for non-critical data)
max.in.flight.requests.per.connection=5
 
# Producer - High Reliability Mode
acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
retries=2147483647
delivery.timeout.ms=120000  # 2 minute total delivery timeout
 
# Consumer - High Throughput Mode
fetch.min.bytes=65536       # 64KB - fetch larger chunks
fetch.max.wait.ms=500       # Wait up to 500ms for fetch.min.bytes
max.poll.records=500        # Process 500 messages per poll
enable.auto.commit=false    # Always manual commit in production
session.timeout.ms=45000    # Avoid unnecessary rebalances
 
# Broker
num.io.threads=16           # I/O threads - match to disk count or cores
num.network.threads=8       # Network threads - match to network cores
socket.send.buffer.bytes=1048576     # 1MB
socket.receive.buffer.bytes=1048576  # 1MB
socket.request.max.bytes=104857600   # 100MB max request size
log.retention.check.interval.ms=300000  # Check retention every 5 minutes

# Connection and channel settings
channel_max = 128           # Max channels per connection
tcp_listen_backlog = 128    # TCP connection queue depth
heartbeat = 60              # Seconds between heartbeats
 
# Memory and disk settings
vm_memory_high_watermark.relative = 0.4   # Throttle at 40% RAM
vm_memory_high_watermark_paging_ratio = 0.5  # Page to disk at 50% of watermark
disk_free_limit.absolute = 2GB  # Minimum free disk space
 
# Performance settings
consumer_timeout = 1800000  # 30 minutes max for consumer to ACK
max_message_size = 134217728  # 128MB max message size

// Load test producer
@Component
public class LoadTestProducer {
 
    public void runLoadTest(int targetTps, Duration duration) {
        RateLimiter rateLimiter = RateLimiter.create(targetTps);
        Instant endTime = Instant.now().plus(duration);
 
        long messagesSent = 0;
        while (Instant.now().isBefore(endTime)) {
            rateLimiter.acquire();
 
            String key = "user-" + ThreadLocalRandom.current().nextInt(1000);
            kafkaTemplate.send("load-test-topic", key, generateTestMessage());
            messagesSent++;
        }
 
        log.info("Load test complete: {} messages sent at {} TPS", messagesSent, targetTps);
    }
}