Cheat SheetsApache KafkaConsumers

Consumers — Cheat Sheet

Apache Kafka · 9 topics. Download the PDF or the Instagram carousel and share it.

Cheat Sheet · AiCanCode.org

Consumers

Apache Kafka9 topicsQuick revision reference

Consumer Groups

Consumers in the same group cooperatively consume partitions; each partition is assigned to exactly one consumer, enabling parallel horizontal scaling up to the partition count.

✓Each partition is assigned to exactly one consumer within a group — this prevents duplicate processing within the group.
✓The maximum parallelism for a consumer group equals the number of partitions; extra consumers sit idle.
✓Multiple consumer groups read the same topic independently — each group gets its own copy of every event (pub-sub fan-out).
✓If a consumer crashes, Kafka triggers a rebalance and reassigns its partitions to surviving group members.
✓CooperativeStickyAssignor performs incremental rebalances so unaffected consumers keep processing during partition handoff.
✓Monitor consumer lag (LOG-END-OFFSET minus CURRENT-OFFSET) per partition — high lag means the consumer cannot keep up.

Java — Spring Boot Kafka

// Spring Boot — @KafkaListener with consumer group
@Component
public class OrderConsumer {

    // All instances of this service share group "order-service"
    // Kafka assigns partitions across them automatically
    @KafkaListener(
        topics   = "orders",
        groupId  = "order-service",
        concurrency = "3"  // 3 listener threads per instance
    )
    public void consume(ConsumerRecord<String, String> record) {
        log.info("Partition={} Offset={} Key={} Value={}",
            record.partition(), record.offset(),
            record.key(), record.value());
        processOrder(record.value());
    }
}

// application.yml
spring:
  kafka:
    consumer:
      group-id: order-service
      auto-offset-reset: earliest   # start from beginning if no committed offset
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

Consumer Offsets

Consumer groups commit offsets to the __consumer_offsets internal topic; on restart or rebalance, the consumer resumes from the last committed offset.

✓Offsets are committed per consumer group per partition to __consumer_offsets internal topic
✓Auto-commit can lose messages: offset committed before processing completes
✓Manual commit (MANUAL_IMMEDIATE) ensures offset committed only after successful processing
✓auto.offset.reset=earliest: start from beginning; latest: start from now (for new groups)
✓Reset offsets with kafka-consumer-groups.sh --reset-offsets to replay historical events
✓At-least-once + idempotent processing is the recommended pattern for most services

Kafka — auto-commit vs manual ack modes

# Auto-commit (default) — simple but can lose messages
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=5000   # commit every 5s

# Manual commit — at-least-once guarantee
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=manual_immediate   # Spring Kafka ack mode

@KafkaListener(topics = "orders")
public void process(Order order, Acknowledgment ack) {
    try {
        orderService.process(order);
        ack.acknowledge();           // commit offset AFTER successful processing
    } catch (RecoverableException e) {
        // Don't ack — record will be redelivered on restart/rebalance
        throw e;
    }
}

# Spring Kafka ack modes:
# RECORD      — commit after each record
# BATCH       — commit after each batch returned by poll()
# MANUAL      — commit when Acknowledgment.acknowledge() is called on batch end
# MANUAL_IMMEDIATE — commit synchronously right when acknowledge() is called

Auto vs Manual Offset Commit

enable.auto.commit=true commits periodically and risks data loss; manual commitSync/commitAsync after processing guarantees at-least-once but requires idempotent consumers.

✓Auto-commit (default) commits offsets on a timer, independently of processing — a crash between poll and commit causes data loss.
✓Disable auto-commit and commit manually after successful processing for at-least-once delivery semantics.
✓commitAsync() is higher throughput; commitSync() is reliable — use async in the loop, sync in the shutdown/finally block.
✓Spring Kafka ack-mode: MANUAL_IMMEDIATE commits on ack.acknowledge(); MANUAL batches to the next poll().
✓Manual commit with at-least-once means duplicates are possible on retry — consumers must be idempotent.
✓Never commit offsets inside the catch block of a transient error — let the message be redelivered.

YAML + Java — Auto-Commit Risk

// application.yml — auto-commit (default, NOT recommended for production)
spring:
  kafka:
    consumer:
      enable-auto-commit: true
      auto-commit-interval: 5000  # commit every 5 seconds regardless of processing
      auto-offset-reset: earliest

// The danger in code
@KafkaListener(topics = "payments")
public void process(List<ConsumerRecord<String, Payment>> records) {
    for (ConsumerRecord<String, Payment> rec : records) {
        // If we crash here, offsets may already be committed — DATA LOSS!
        paymentService.handle(rec.value());
    }
}

Consumer Rebalancing

When a consumer joins or leaves a group, the broker-coordinated rebalance reassigns partitions; use Cooperative Sticky Assignor (incremental rebalance) to minimise disruption.

✓Rebalance triggered by: consumer join/leave, heartbeat timeout, max.poll.interval.ms exceeded
✓Eager rebalance: all consumers pause and revoke all partitions — causes processing gaps
✓Cooperative Sticky Assignor: only revoked partitions pause; rest continue — default in Kafka 3.x
✓Tune max.poll.records to prevent exceeding max.poll.interval.ms on slow processing
✓ConsumerRebalanceListener.onPartitionsRevoked(): commit offsets before revocation to avoid duplicates
✓heartbeat.interval.ms should be < 1/3 of session.timeout.ms

Kafka — rebalance timeout tuning

# Key timeouts — tuned to reduce spurious rebalances
session.timeout.ms=45000         # max time broker waits for heartbeat (default: 45s)
heartbeat.interval.ms=3000       # how often consumer sends heartbeat (should be <1/3 session)
max.poll.interval.ms=300000      # max time between poll() calls before consumer is dead

# Spring Kafka equivalent
spring.kafka.consumer.properties.session.timeout.ms=45000
spring.kafka.consumer.properties.max.poll.interval.ms=300000
spring.kafka.listener.poll-timeout=3000

# Reduce max.poll.records if processing is slow — prevents exceeding max.poll.interval.ms
spring.kafka.consumer.max-poll-records=100    # default 500

# Warning signs of rebalance thrashing:
# - Logs: "Attempt to heartbeat failed since group is rebalancing"
# - Consumer lag spikes at regular intervals
# - Duplicate records processed (offset re-committed after rebalance)

Consumer Poll Loop

The consumer calls poll() in a loop to heartbeat with the coordinator and fetch records; max.poll.interval.ms must exceed the time to process a batch or the consumer is evicted.

✓poll() serves double duty: fetches records AND sends the consumer heartbeat to the group coordinator.
✓max.poll.interval.ms is the max time between poll() calls; exceed it and the coordinator evicts the consumer, triggering a rebalance.
✓Reduce max.poll.records to ensure each batch processes within max.poll.interval.ms.
✓session.timeout.ms controls the background heartbeat timeout (GC pauses / JVM unresponsiveness); max.poll.interval.ms controls processing timeout.
✓Always commit offsets in onPartitionsRevoked to avoid re-processing during planned rebalances.
✓Spring Kafka's @KafkaListener manages the poll loop automatically — configure max-poll-records and max.poll.interval.ms in application.yml.

Java — 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();

Consumer Lag

Lag is the difference between the latest produced offset and the last committed consumer offset; high lag indicates the consumer cannot keep up with the producer rate.

✓Consumer lag = log-end-offset − current-committed-offset per partition; monitor per-partition, not just total.
✓A steadily growing lag is more alarming than a high-but-stable lag — alert on lag velocity, not just threshold.
✓Root causes: too few consumers, slow processing, max.poll.records too high causing poll-interval breach, downstream bottleneck.
✓Max consumers in a group = partition count; adding more consumers beyond that has no effect — they sit idle.
✓Use kafka-consumer-groups.sh or AdminClient.listConsumerGroupOffsets() to measure lag programmatically.
✓Prometheus Kafka Exporter / JMX Exporter exposes kafka_consumer_group_lag metric for Grafana dashboards and alerting.

Shell + Java — Measuring Lag

# CLI — describe a consumer group to see lag per partition
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe \
    --group order-processing-service

# Output:
# GROUP                     TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processing-service  orders  0          5000            5000            0    ← healthy
# order-processing-service  orders  1          4800            5200            400  ← lagging!
# order-processing-service  orders  2          5100            5100            0

// Programmatic lag calculation
AdminClient admin = AdminClient.create(Map.of(
    AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"
));

// Get committed offsets for the group
Map<TopicPartition, OffsetAndMetadata> committed =
    admin.listConsumerGroupOffsets("order-processing-service")
         .partitionsToOffsetAndMetadata().get();

// Get end offsets (latest produced)
Map<TopicPartition, Long> endOffsets =
    admin.listOffsets(
        committed.keySet().stream()
            .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())))
         .all().get()
         .entrySet().stream()
         .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

// Calculate lag per partition
committed.forEach((tp, meta) -> {
    long lag = endOffsets.get(tp) - meta.offset();
    System.out.printf("Partition %s: lag = %d%n", tp, lag);
});

Consumer Rebalancing

Rebalancing redistributes partition assignments among consumers in a group. It is triggered by group membership changes and can cause brief processing pauses.

✓Eager rebalance: ALL consumers pause — all partitions revoked and reassigned
✓Cooperative Sticky: only affected partitions move, others continue
✓max.poll.interval.ms exceeded → consumer kicked from the group
✓Implement ConsumerRebalanceListener to commit offsets before revocation
✓Static group membership (group.instance.id) eliminates rebalances on rolling restarts

Java — CooperativeStickyAssignor

// Use Cooperative Sticky — best for most use cases
spring.kafka.consumer.properties.partition.assignment.strategy=\
  org.apache.kafka.clients.consumer.CooperativeStickyAssignor

@Component
public class RebalanceListener implements ConsumerRebalanceListener {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync(currentOffsets(partitions)); // commit before revoke
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Assigned: {}", partitions);
    }
}

Consumer Lag Monitoring

Consumer lag is the gap between the latest produced offset and the committed consumer offset. High lag means consumers are falling behind and SLA breaches are imminent.

✓Lag = log-end-offset minus committed-offset per partition
✓Growing lag → consumer is slower than the producer rate
✓Lag exceeding retention period → consumer will miss messages
✓Export lag to Prometheus via kafka-lag-exporter
✓Alert when lag > threshold AND is growing (not just a spike)

Kafka — lag monitoring

# Check lag for a consumer group
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-processor --describe

# Programmatic lag check
AdminClient admin = AdminClient.create(props);
ListConsumerGroupOffsetsResult offsets =
    admin.listConsumerGroupOffsets("order-processor");
Map<TopicPartition, OffsetAndMetadata> committed =
    offsets.partitionsToOffsetAndMetadata().get();
// compare committed offset vs log-end-offset for each partition

Dead Letter Topics (DLT)

A Dead Letter Topic stores messages that a consumer has failed to process after all retry attempts. It enables non-blocking error handling and deferred reprocessing.

✓DLT prevents a poison-pill message from blocking the entire partition
✓Naming convention: {topic}.DLT; same partition preserves ordering
✓DeadLetterPublishingRecoverer routes to DLT after all retries exhausted
✓DLT headers contain exception class, message, and stack trace
✓Monitor DLT topic size — growing DLT signals repeated failures

Spring Kafka — DLT with DefaultErrorHandler

@Configuration
public class KafkaConfig {
    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(template,
                (r, e) -> new TopicPartition(r.topic() + ".DLT", r.partition()));
        BackOff backOff = new FixedBackOff(2000L, 3L); // 3 retries, 2s gap
        return new DefaultErrorHandler(recoverer, backOff);
    }
}

@KafkaListener(topics = "orders.DLT", groupId = "dlt-processor")
public void handleDlt(ConsumerRecord<String, Order> record,
        @Header(KafkaHeaders.EXCEPTION_MESSAGE) String errorMsg) {
    log.error("DLT: key={} error={}", record.key(), errorMsg);
}

Learn this free with Aria, your AI tutor → AiCanCode.org/learn/kafka

Cheat Sheet

Consumers

Apache Kafka

9 topics · swipe →

AiCanCode.org

Consumers

Consumer Groups

✓Each partition is assigned to exactly one consumer within a group — this prevents duplicate processing within the group.
✓The maximum parallelism for a consumer group equals the number of partitions; extra consumers sit idle.
✓Multiple consumer groups read the same topic independently — each group gets its own copy of every event (pub-sub fan-out).
✓If a consumer crashes, Kafka triggers a rebalance and reassigns its partitions to surviving group members.
✓CooperativeStickyAssignor performs incremental rebalances so unaffected consumers keep processing during partition handoff.

// Spring Boot — @KafkaListener with consumer group
@Component
public class OrderConsumer {

    // All instances of this service share group "order-service"
    // Kafka assigns partitions across them automatically
    @KafkaListener(
        topics   = "orders",
        groupId  = "order-service",
…

AiCanCode.org

Consumers

Consumer Offsets

✓Offsets are committed per consumer group per partition to __consumer_offsets internal topic
✓Auto-commit can lose messages: offset committed before processing completes
✓Manual commit (MANUAL_IMMEDIATE) ensures offset committed only after successful processing
✓auto.offset.reset=earliest: start from beginning; latest: start from now (for new groups)
✓Reset offsets with kafka-consumer-groups.sh --reset-offsets to replay historical events

# Auto-commit (default) — simple but can lose messages
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=5000   # commit every 5s

# Manual commit — at-least-once guarantee
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=manual_immediate   # Spring Kafka ack mode

@KafkaListener(topics = "orders")
…

AiCanCode.org

Consumers

Auto vs Manual Offset Commit

✓Auto-commit (default) commits offsets on a timer, independently of processing — a crash between poll and commit causes data loss.
✓Disable auto-commit and commit manually after successful processing for at-least-once delivery semantics.
✓commitAsync() is higher throughput; commitSync() is reliable — use async in the loop, sync in the shutdown/finally block.
✓Spring Kafka ack-mode: MANUAL_IMMEDIATE commits on ack.acknowledge(); MANUAL batches to the next poll().
✓Manual commit with at-least-once means duplicates are possible on retry — consumers must be idempotent.

// application.yml — auto-commit (default, NOT recommended for production)
spring:
  kafka:
    consumer:
      enable-auto-commit: true
      auto-commit-interval: 5000  # commit every 5 seconds regardless of processing
      auto-offset-reset: earliest

// The danger in code
…

AiCanCode.org

Consumers

Consumer Rebalancing

✓Rebalance triggered by: consumer join/leave, heartbeat timeout, max.poll.interval.ms exceeded
✓Eager rebalance: all consumers pause and revoke all partitions — causes processing gaps
✓Cooperative Sticky Assignor: only revoked partitions pause; rest continue — default in Kafka 3.x
✓Tune max.poll.records to prevent exceeding max.poll.interval.ms on slow processing
✓ConsumerRebalanceListener.onPartitionsRevoked(): commit offsets before revocation to avoid duplicates

# Key timeouts — tuned to reduce spurious rebalances
session.timeout.ms=45000         # max time broker waits for heartbeat (default: 45s)
heartbeat.interval.ms=3000       # how often consumer sends heartbeat (should be <1/3 session)
max.poll.interval.ms=300000      # max time between poll() calls before consumer is dead

# Spring Kafka equivalent
spring.kafka.consumer.properties.session.timeout.ms=45000
spring.kafka.consumer.properties.max.poll.interval.ms=300000
spring.kafka.listener.poll-timeout=3000
…

AiCanCode.org

Consumers

Consumer Poll Loop

✓poll() serves double duty: fetches records AND sends the consumer heartbeat to the group coordinator.
✓max.poll.interval.ms is the max time between poll() calls; exceed it and the coordinator evicts the consumer, triggering a rebalance.
✓Reduce max.poll.records to ensure each batch processes within max.poll.interval.ms.
✓session.timeout.ms controls the background heartbeat timeout (GC pauses / JVM unresponsiveness); max.poll.interval.ms controls processing timeout.
✓Always commit offsets in onPartitionsRevoked to avoid re-processing during planned rebalances.

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) {
…

AiCanCode.org

Consumers

Consumer Lag

✓Consumer lag = log-end-offset − current-committed-offset per partition; monitor per-partition, not just total.
✓A steadily growing lag is more alarming than a high-but-stable lag — alert on lag velocity, not just threshold.
✓Root causes: too few consumers, slow processing, max.poll.records too high causing poll-interval breach, downstream bottleneck.
✓Max consumers in a group = partition count; adding more consumers beyond that has no effect — they sit idle.
✓Use kafka-consumer-groups.sh or AdminClient.listConsumerGroupOffsets() to measure lag programmatically.

# CLI — describe a consumer group to see lag per partition
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe \
    --group order-processing-service

# Output:
# GROUP                     TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processing-service  orders  0          5000            5000            0    ← healthy
…

AiCanCode.org

Consumers

Consumer Rebalancing

✓Eager rebalance: ALL consumers pause — all partitions revoked and reassigned
✓Cooperative Sticky: only affected partitions move, others continue
✓max.poll.interval.ms exceeded → consumer kicked from the group
✓Implement ConsumerRebalanceListener to commit offsets before revocation
✓Static group membership (group.instance.id) eliminates rebalances on rolling restarts

// Use Cooperative Sticky — best for most use cases
spring.kafka.consumer.properties.partition.assignment.strategy=\
  org.apache.kafka.clients.consumer.CooperativeStickyAssignor

@Component
public class RebalanceListener implements ConsumerRebalanceListener {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync(currentOffsets(partitions)); // commit before revoke
…

AiCanCode.org

Consumers

Consumer Lag Monitoring

✓Lag = log-end-offset minus committed-offset per partition
✓Growing lag → consumer is slower than the producer rate
✓Lag exceeding retention period → consumer will miss messages
✓Export lag to Prometheus via kafka-lag-exporter
✓Alert when lag > threshold AND is growing (not just a spike)

# Check lag for a consumer group
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-processor --describe

# Programmatic lag check
AdminClient admin = AdminClient.create(props);
ListConsumerGroupOffsetsResult offsets =
    admin.listConsumerGroupOffsets("order-processor");
Map<TopicPartition, OffsetAndMetadata> committed =
…

AiCanCode.org

Consumers

Dead Letter Topics (DLT)

✓DLT prevents a poison-pill message from blocking the entire partition
✓Naming convention: {topic}.DLT; same partition preserves ordering
✓DeadLetterPublishingRecoverer routes to DLT after all retries exhausted
✓DLT headers contain exception class, message, and stack trace
✓Monitor DLT topic size — growing DLT signals repeated failures

@Configuration
public class KafkaConfig {
    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(template,
                (r, e) -> new TopicPartition(r.topic() + ".DLT", r.partition()));
        BackOff backOff = new FixedBackOff(2000L, 3L); // 3 retries, 2s gap
        return new DefaultErrorHandler(recoverer, backOff);
…

AiCanCode.org

Master Consumers

Full lessons + an AI tutor, free.

AiCanCode.org

/learn/kafka