Learn Apache Kafka - Consumers: Reading Data from Kafka

#Introduction

If the producer is the write side, the consumer is the read side — and this is where many common mistakes happen: consumers reading the wrong offset, commits happening before data finishes processing, or rebalance storms that destabilize the cluster.

Episode 6 takes you to mastery of consumers: from subscription and polling, the consumer lifecycle, consumer group mechanics and rebalancing with various assignment strategies, to correct offset management. You'll also learn to tune parameters like fetch.min.bytes, max.poll.records, and session.timeout.ms.

Correct consumer patterns are the key to a reliable, easily scalable system. Let's dissect them one by one.

#Consumer Fundamentals

#Configuration and Subscription

Consumers need bootstrap.servers, deserializers, and a group.id to form a group. Subscription is done with subscribe() to one or more topics:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processor");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> r : records) {
            System.out.println(r.key() + " -> " + r.value());
        }
    }
}

#Polling and ConsumerRecord

poll() is the heart of the consumer: it blocks for up to the given timeout to fetch a batch of records, while also sending heartbeats to the group coordinator. Each ConsumerRecord carries a topic, partition, offset, key, value, and timestamp. The while (true) loop is the standard pattern — one thread, one consumer.

#Consumer Lifecycle

Consumers go through several phases: assign partitions (join group), read records from the stored offset, commit, and finally close(), which releases partitions and sends a final commit. Failing to close a consumer properly can hold partitions until the session timeout.

#Consumer Groups

#Group Coordination

Every group has a group coordinator — one of the brokers — that tracks group members and partition assignments. When members join or leave, the coordinator triggers a rebalance: stops all members from consuming, recomputes assignments, and resumes. This is transparent but expensive, so it needs to be minimized.

#Partition Assignment Strategies

The assignment strategy determines how partitions are divided among members:

• Range: divides partitions per topic sequentially; can create imbalance across many topics.
• RoundRobin: distributes partitions of all topics in turn, giving more even results.
• Sticky: preserves the previous assignment as much as possible during rebalance.
• CooperativeSticky (3.1+): incremental rebalance — only the partitions that changed are moved, without full stop-the-world.

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

#Rebalancing and Static Membership

A full (eager) rebalance briefly stops all members. CooperativeSticky avoids this, and static membership (group.instance.id) keeps a consumer considered a member even through short restarts, preventing rebalances during rolling deployments. Combining both keeps groups stable in production.

#Offset Management

#Auto-Commit vs Manual Commit

By default enable.auto.commit=true and auto.commit.interval.ms=5000: offsets are committed automatically every 5 seconds. The risk is at-least-once — if the application crashes between processing and commit, records will be reprocessed. For full control, disable auto-commit and commit manually after processing finishes:

enable.auto.commit=false

for (ConsumerRecord<String, String> r : records) {
    process(r);
}
consumer.commitSync();

commitSync() blocks until the commit succeeds; commitAsync() returns the result via a callback and doesn't block the loop. The best pattern: commitAsync() in the normal loop, then commitSync() at close as a final safety net.

#Offset Storage and Seeking

Consumer offsets are stored in the internal __consumer_offsets topic, managed by the group coordinator. You can ignore the stored position and read from the beginning (seekToBeginning), the end (seekToEnd), or a specific offset. auto.offset.reset determines the behavior when a group has no offset yet:

• earliest: read from the beginning of the partition.
• latest: read only new records (default).
• none: error if the offset is not found.

#Consumer Configuration Tuning

#Fetch Parameters

fetch.min.bytes and fetch.max.wait.ms balance throughput and latency: the broker holds the response until at least N bytes accumulate or waits at most M ms. max.poll.records limits the number of records per poll() — important for keeping processing time within max.poll.interval.ms.

#Timeout and Heartbeat

• session.timeout.ms (default 45s): the broker considers a consumer dead if no heartbeat arrives within this window.
• heartbeat.interval.ms: heartbeat frequency, usually a third of the session timeout.
• max.poll.interval.ms (default 5 minutes): the total time limit for processing the results of one poll(); if exceeded, the consumer is evicted from the group.

#Advanced Consumer Patterns

Some frequently used patterns:

• One consumer per partition for maximum ordering and isolation.
• Consumer groups for horizontal parallelism; add members up to the number of partitions.
• Manual partition assignment with assign() when you don't want the coordinator to decide.
• Pause/resume (pause() and resume()) for backpressure when downstream slows down.

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-processor --describe

The output shows per partition: current-offset, log-end-offset, lag, and the consumer holding the partition. kafka-consumer-groups.sh --describe will become your routine tool for monitoring consumption progress.

#Closing

In this episode 6 you've mastered consumers: subscription and polling, lifecycle, consumer groups with the Range, RoundRobin, Sticky, and CooperativeSticky assignment strategies, offset management with auto-commit and manual commit, plus tuning of fetch.min.bytes, max.poll.records, and timeout parameters.

The key takeaways:

• poll() simultaneously fetches records and sends heartbeats.
• One partition is read by only one consumer in a group; add consumers up to the partition count.
• CooperativeSticky and static membership minimize rebalance impact.
• Commit offsets after processing finishes; use commitAsync plus commitSync at close.
• Offsets are stored in the __consumer_offsets topic.
• auto.offset.reset=earliest for reading from the beginning, latest for only new records.

In the next episode 7 we'll prepare the data being sent: message serialization and schema management — from String, JSON, Avro, and Protobuf, to the role of the Confluent Schema Registry in schema evolution and compatibility types. You'll learn to define Avro schemas and choose the right format for your application!