Consumer Poll Loop

KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Commit offsets for revoked partitions before rebalance completes
        consumer.commitSync();
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Assigned partitions: {}", partitions);
    }
});

boolean running = true;
while (running) {
    // poll() also sends heartbeat + triggers rebalance callbacks
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(500));

    for (ConsumerRecord<String, Order> record : records) {
        processOrder(record.value());
    }

    consumer.commitAsync();  // async offset commit after each batch
}

consumer.commitSync();  // final synchronous commit on shutdown
consumer.close();

# Tuning for slow processing

# Option 1: Reduce batch size — fewer records per poll, faster to process
spring:
  kafka:
    consumer:
      max-poll-records: 50          # default 500; lower for slow processing
      properties:
        max.poll.interval.ms: 60000 # 60s — must exceed worst-case batch processing time

# Option 2: Increase the timeout to match processing reality
spring:
  kafka:
    consumer:
      properties:
        max.poll.interval.ms: 300000  # 5 min (default) — increase if needed

# Monitoring: if you see frequent unexpected rebalances in logs, check:
# 1. How long each batch is taking to process
# 2. Whether max.poll.interval.ms < actual processing time

// Spring Kafka — pause the listener container for heavy processing
// and resume after — keeps the poll loop running while processing offline
@KafkaListener(topics = "orders", id = "orderListener")
public void onOrder(List<Order> orders) {
    registry.getListenerContainer("orderListener").pause();
    // heavy async processing — poll loop still runs (just pauses fetch)
    processAsync(orders)
        .thenRun(() -> registry.getListenerContainer("orderListener").resume());
}

spring:
  kafka:
    consumer:
      properties:
        # Background heartbeat timeout — increase if you have long GC pauses
        session.timeout.ms: 45000      # default; increase on JVM with large heaps

        # Poll interval — increase beyond your worst-case batch processing time
        max.poll.interval.ms: 120000   # 2 minutes

        # Heartbeat interval — should be 1/3 of session.timeout.ms
        heartbeat.interval.ms: 15000   # sends heartbeat every 15s