Kafka

60-Day Kafka 4 Learning Plan · Week 1 — Day 6 of 60

Editorial Team

60-Day Kafka 4 Learning Plan · Week 1 — Foundations · Spring Boot Integration Sources: Kafka: The Definitive Guide Ch.4 · kafka.apache.org/43 · Spring Kafka Docs

Goal

Understand consumer groups, the poll loop, offset commit strategies, rebalancing, and implement @KafkaListener patterns in Spring Boot 3 — plus error recovery, lag monitoring, and testing.

1. Consumer Groups

Each consumer in a group is assigned a subset of partitions. The broker coordinates assignment via the Group Coordinator.

Scenario A — 2 consumers, 4 partitions (balanced)
  P0 P1 → C1
  P2 P3 → C2

Scenario B — 4 consumers, 4 partitions (★ ideal)
  P0 → C1  |  P1 → C2  |  P2 → C3  |  P3 → C4

Scenario C — 5 consumers, 4 partitions (⚠ waste)
  P0 → C1  |  P1 → C2  |  P2 → C3  |  P3 → C4  |  C5 → idle

Golden rule: num_consumers ≤ num_partitions. Each consumer group receives all messages independently — multiple groups on the same topic each get a full copy.

2. Poll Loop & Internals

consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, Order> r : records) {
        process(r);
    }
    consumer.commitAsync();
}

How it works:

  • poll() fetches records and sends heartbeats to the broker
  • session.timeout.ms (default 45s): missing heartbeat window before consumer is declared dead → rebalance
  • max.poll.interval.ms (default 5min): too slow between polls → also triggers rebalance
  • Always call consumer.close() on shutdown to trigger clean rebalance instead of waiting for timeout

Heartbeat vs poll interval: since Kafka 0.10.1, heartbeats run on a separate background thread, decoupled from poll(). This means a slow process() call won’t kill the session as long as heartbeats keep flowing — but it will eventually breach max.poll.interval.ms if processing takes too long, triggering a rebalance anyway. Both timeouts matter, for different failure modes.

3. Offset Commit Modes

Auto Commit

enable.auto.commit: true
auto.commit.interval.ms: 5000
  • Easiest but ⚠ risk of duplicates if consumer crashes between commit intervals

Sync Commit

consumer.commitSync();  // blocks until broker ACKs, retries on failure
  • Safe but slower — use for exactly-once requirements
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.error("Commit failed for offsets {}", offsets, exception);
    }
});
  • Non-blocking with error visibility
  • Pair with a final commitSync() on consumer close

Spring Boot AckMode

At-least-once vs exactly-once: committing after processing (the pattern above) gives at-least-once delivery — a crash between processing and commit causes reprocessing on restart. True exactly-once requires either idempotent downstream writes (dedupe by message key) or a transactional consume-transform-produce pipeline (§9).

4. Rebalancing

Triggers:

  • Consumer joins or leaves the group
  • Consumer crashes (heartbeat timeout exceeded)
  • New partitions added to a topic

Eager Rebalance (classic — stop-the-world)

1. All consumers revoke their partitions
2. Group Leader proposes new assignment
3. All consumers rejoin and accept assignment
4. Consuming resumes

⚠ Full pause during rebalance — no messages processed.

Cooperative Rebalance (Kafka 2.4+ — incremental)

Only the partitions being moved are revoked.
Other partitions keep being consumed — no global pause.

Enable with: partition.assignment.strategy=CooperativeStickyAssignor

Static Membership — avoid unnecessary rebalances

For rolling deploys/restarts, a consumer leaving and rejoining within session.timeout.ms still triggers a rebalance by default. Static membership fixes this by giving each consumer instance a stable identity:

spring.kafka.consumer:
  properties:
    group.instance.id: order-svc-pod-${HOSTNAME}  # stable per instance

With a group.instance.id set, a brief disconnect (pod restart, network blip) is treated as the same member returning — no rebalance is triggered as long as it rejoins within session.timeout.ms. Essential for Kubernetes rolling deploys with many partitions.

5. Spring Boot Config & @KafkaListener

application.yml

spring:
  kafka:
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
    listener:
      ack-mode: BATCH

Single Record Listener

@KafkaListener(topics = "orders", groupId = "order-svc")
public void consume(ConsumerRecord<String, Order> record) {
    log.info("Received order={} at offset={}", record.value(), record.offset());
    // process...
}

Batch Listener

@KafkaListener(topics = "events", containerFactory = "batchFactory")
public void consumeBatch(List<Event> events, Acknowledgment ack) {
    events.forEach(e -> process(e));
    ack.acknowledge();  // manual commit after batch
}

ConcurrentKafkaListenerContainerFactory (for batch mode)

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Event> batchFactory(
        ConsumerFactory<String, Event> cf) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, Event>();
    factory.setConsumerFactory(cf);
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
    return factory;
}

6. Error handling — retries, poison pills, and DLQ

A poison pill is a message that always fails processing (bad JSON, unexpected schema, business rule violation). Without a strategy, it blocks the partition forever — the consumer keeps retrying the same offset.

DefaultErrorHandler + dead-letter topic

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
    var recoverer = new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));

    // retry 3 times, 1s apart, then send to DLT
    var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3));

    // don't retry business/deserialization errors — straight to DLT
    handler.addNotRetryableExceptions(DeserializationException.class, IllegalArgumentException.class);
    return handler;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
        ConsumerFactory<String, Order> cf, DefaultErrorHandler errorHandler) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, Order>();
    factory.setConsumerFactory(cf);
    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

Deserialization poison pills need special handling since the exception happens before your listener code runs — use ErrorHandlingDeserializer as a wrapper:

spring.kafka.consumer:
  key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
  value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
  properties:
    spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
    spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

This lets the broken record flow into the error handler (and DLT) instead of crashing the container on startup.

7. Monitoring consumer lag

Consumer lag = latest offset in the partition − last committed offset. It’s the single most important consumer health metric — rising lag means the consumer can’t keep up with the producer.

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc

For dashboards, expose lag via Micrometer:

@Bean
public MeterRegistryCustomizer<MeterRegistry> kafkaConsumerMetrics(
        ConcurrentKafkaListenerContainerFactory<?, ?> factory) {
    return registry -> factory.getContainerProperties()
        .setMicrometerEnabled(true); // auto-exposes kafka.consumer.* metrics
}

Rule of thumb: lag that grows faster than it drains during peak traffic means you’re under-provisioned on partitions/consumers, not just experiencing a temporary blip.

8. Multi-threading models

factory.setConcurrency(3); // 3 threads, each consuming a share of the partitions

concurrency should never exceed the topic’s partition count — extra threads sit idle (same waste as Scenario C in §1).

9. Exactly-once consume-transform-produce

When a consumer reads a message, transforms it, and produces a new message — and both must succeed or fail together — combine a transactional producer (Day 5, §4) with the consumer’s offset commit:

kafkaTemplate.executeInTransaction(ops -> {
    ops.sendOffsetsToTransaction(currentOffsets, consumerGroupMetadata); // commit offset AS PART of the transaction
    ops.send("enriched-orders", key, enrichedPayload);
    return true;
});

This way, the input offset commit and the output message are atomic — a crash mid-way means neither happened, so on restart the same input record is reprocessed cleanly (no half-applied state). Downstream consumers must use isolation.level=read_committed to only see the finished result.

10. Testing consumers

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = "orders")
class ConsumerIntegrationTest {

    @Autowired KafkaTemplate<String, Order> kafkaTemplate;

    @SpyBean OrderListener orderListener; // the @KafkaListener bean

    @Test
    void consumesAndProcessesOrder() {
        kafkaTemplate.send("orders", "key1", new Order("123"));

        await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
            verify(orderListener).consume(any())
        );
    }
}

Use awaitility (await().untilAsserted(...)) instead of Thread.sleep() — consumption is asynchronous, and polling for the assertion avoids flaky, slow tests.

11. Common pitfalls

  • Long processing inside the listener without adjusting max.poll.interval.ms — triggers unwanted rebalances mid-processing
  • Auto-commit with slow processing — offset commits on a timer, not tied to actual completion, risking data loss on crash
  • Ignoring poison pills — one bad message with no DLT strategy can permanently stall a partition
  • concurrency > partition count — wasted threads that never receive work
  • Not setting group.instance.id in Kubernetes — every rolling deploy causes a full rebalance storm across the whole group
  • Mixing manual and auto commit — e.g. calling ack.acknowledge() while enable-auto-commit: true is still set, causing confusing double-commit behavior

12. Key Consumer Configs

Key Takeaways

  • Consumer groups allow parallel processing — scale consumers up to num_partitions max
  • Offset commits track progress — use BATCH AckMode in Spring Boot with enable-auto-commit: false
  • Rebalances redistribute partitions when group membership changes — Cooperative Rebalance (Kafka 2.4+) and static membership both minimize disruption
  • Poison pills need a DefaultErrorHandler + DLT strategy, or they’ll stall a partition indefinitely
  • Consumer lag is the primary health signal — monitor records-lag-max continuously, not just on incidents
  • True exactly-once needs transactional offset commits (sendOffsetsToTransaction), not just idempotent producers
  • Always close consumers cleanly to avoid waiting for session.timeout.ms

Support me through GitHub Sponsors.

Next

➡️ Day 7 — First Project: Spring Boot Kafka Lab — producer + consumer end-to-end

Resources

👉 Link to Medium blog

, ,

Related Posts

Kafka

60-Day Kafka 4 Learning Plan · Week 3 — Day 16 of 60

Editorial Team
Kafka

60-Day Kafka 4 Learning Plan · Week 3 — Day 15 of 60

Editorial Team