Kafka Consumers — Full Deep Dive (basic → advanced)

Basic concepts: consumers, consumer groups, and partitions

In Kafka, A Topic is an ordered log of messages split into partitions.

Partition is an append-only sequence of records, each record has a unique offset (0,1,2…)

Consumer is a client that reads records from one or more partitions.

Consumer Group is a logical group of consumers (all sharing the same group.id) that cooperatively consume a topic. Each partition in the topic is assigned to exactly one consumer within the same group. Multiple consumer groups can read the same topic independently (fan-out).

Order is guaranteed per partition, not across partitions. If number of partitions < number of consumers, then some consumers are idle. If number of partitions are greater than number of consumers then some consumers read multiple partitions.

Offsets — What they are and why they matter

Message Offset or Log Offset : Lets first understand Message Offset or Log Offset. Every record written to a partition in Kafka is assigned a unique offset, which is a sequential, monotonically increasing 64-bit integer. Offsets start at 0 for the first message and increase by 1 for every new record appended. Offsets are immutable i.e. once assigned to a record, they never change even if you re-read or replay. This offset is owned by Kafka’s log itself, not by consumers.

Consumer Offset (Committed Offset) : Kafka consumers track how far they have read in each partition. That tracking value (the last offset that was successfully processed + 1) is the consumer offset. It is stored in an internal Kafka topic called __cosumer_offsets. When we say “committing an offset”, we are writing this consumer offset to that internal topic. This is a topic like any other: it has partitions and replication. Because it is compacted, only the latest committed offset per (group, topic, partition) is retained permanently.

Let’s say partition 0 has message offsets 0,1,2,3,4. Consumer group group-A has a consumer instance C1 reading this partition. Now C1 process message at message offset 0 and then commit consumer offset 1. That means “I’ve processed all records up to (but not including) offset 1. Next time, I’ll start from 1.” The committed offset = next record to read (not the last processed record).

Offsets are tracked per topic, partition and per consumer group [group.id, topic, partition]. Each consumer group maintains its own committed offset for each topic and its partition.

How offset committing works (auto vs manual)

Auto-Commit — When we call consumer.poll() then assume poll returns messages with offsets 10, 11, 12, 13, 14. Consumer fetches messages from the broker (using fetch.min.bytes, max.poll.records, etc.). The consumer’s internal position (in memory) moves forward to the next offset to fetch → In this case, from 10–14 fetched, next offset = 15. Now we start processing records in our loop. Auto-commit thread (in background) will run, every auto.commit.interval.ms (say, 5000 ms) [provide enable.auto.commit=true], commit the last known internal position = 15 — regardless of whether our app has finished processing those records.

Now If the consumer crashes after committing but before processing, those messages (10–14) are lost forever — because Kafka believes they’re already processed. This is why auto-commit leads to at-most-once semantics in such cases. If consumer crash after processing but before committing then duplicate processing of records happen. This is why auto-commit leads to at-least once semantics as well. We have no guarantee or control over which of these will happen in auto-commit mode.

Manual Commit — when enable.auto.commit=false, we call:

consumer.commitSync() — synchronous, blocks until broker ack; good for strong guarantees.

consumer.commitAsync() — non-blocking; faster but needs error handling for failures.

We typically commit the offset after our successfully process the record(s) i.e. after processing a single record, a batch, or any logical checkpoint. Common patterns:

• Commit per message (low throughput, simple).
• Commit per batch (group of records) — commit offset of the last processed record in batch.
• Commit at logical checkpoints (after saving results to DB, etc.).

Below configurations at consumer side are important to control Kafka consumer fetches data from brokers:

1. max.poll.records — Maximum number of records returned in a single call to poll().
2. max.poll.interval.ms — Maximum delay allowed between two consecutive calls to poll() before Kafka considers the consumer dead (and triggers a rebalance).
3. fetch.min.bytes — The minimum amount of data (in bytes) that the broker should send in a fetch response. Broker will wait until it has at least fetch.min.bytes of data for the consumer before sending it, or until fetch.max.wait.ms expires (default 500 ms).
4. fetch.max.bytes — Maximum amount of data (in bytes) that the broker will return per partition per fetch request.
5. fetch.max.wait.ms — The maximum amount of time (in milliseconds) that the broker will wait before sending data to the consumer, even if fetch.min.bytes has not yet been satisfied.

Delivery semantics: at-most-once, at-least-once, exactly-once

• At-most-once: offsets are committed before processing → messages may be lost if processing fails (no reprocessing).
• At-least-once (most common): offsets are committed after processing → if a crash occurs before commit, records may be processed again (duplicates possible).
• Exactly-once: Kafka supports exactly-once semantics (EOS) in end-to-end scenarios using:

Idempotent producer (enable.idempotence=true) to avoid duplicate writes from producers.

For Kafka->Kafka, By using a transactional producer and sendOffsetsToTransaction(), both offset commits and produced results become part of the same Kafka transaction — committed or rolled back together by the broker.

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

for (ConsumerRecord<String, String> record : records) {
        process(record);
        currentOffsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
        );
    }

// within a transaction:
producer.beginTransaction();
producer.send(new ProducerRecord<>("output-topic", key, value));
producer.sendOffsetsToTransaction(currentOffsets, "my-consumer-group");
producer.commitTransaction();

Kafka consumer liveness — detailed explanation

Kafka uses two mechanisms to detect failed consumers and trigger rebalances:

• Heartbeats (client ↔ group coordinator): are used to tell the coordinator “I’m alive.” It is controlled by heartbeat.interval.ms i.e. how often the client sends heartbeat requests and session.timeout.ms i.e. if coordinator sees no heartbeats within this window → it marks the member dead and triggers a rebalance.
• Poll liveness: max.poll.interval.ms is the maximum time the consumer client allows between consecutive poll() calls before the client considers itself non-responsive and triggers a group rejoin/leave (causing a rebalance). If our processing of a batch takes longer than max.poll.interval.ms, the consumer will be kicked out of the group (rebalance), and partitions will be reassigned. Design implication: Either keep max.poll.records small so each poll processes quickly or use multi-threaded handing but ensure heartbeats are maintained.

Who (or what) is the Group Coordinator?

The Group Coordinator is a special internal role played by one of the Kafka brokers in our cluster. It’s not a separate service — it’s just one of the brokers that is responsible for managing consumer groups. Each consumer group in Kafka is assigned to exactly one broker (called its group coordinator). This coordinator is responsible for:

• Tracks which consumers (clients) have joined or left the group.
• Detects failed consumers using heartbeats and session timeouts.
• Initiates rebalances when membership or topic partition assignments change.
• Coordinates which partitions of each subscribed topic go to which consumer.
• Works with one of the consumers (the group leader) that performs the actual assignment algorithm (range, round-robin, sticky, etc.).
• Stores committed offsets for the group

Kafka uses a hash-based algorithm on the consumer group ID to pick a coordinator broker. So every consumer group has one coordinator, and every broker might act as a coordinator for some subset of groups.

When a consumer starts:

1. It sends a FindCoordinatorRequest to any broker.
2. That broker replies with the address of the correct coordinator.
3. The consumer then talks directly to that coordinator for all JoinGroup, Heartbeat, LeaveGroup, and CommitOffset requests.

What is the Group Leader in Kafka?

The Group Leader is one consumer within a consumer group that is temporarily chosen by the Group Coordinator to perform partition assignment during a rebalance.

It’s a logical role, not a permanent one — the leader can change any time the group rebalances. When all consumers in a group send a JoinGroup request to the Group Coordinator, the first consumer that sends the JoinGroup request (or sometimes the one picked deterministically by the coordinator) is designated as the Group Leader for that rebalance. The Coordinator then:

• Sends the Group Leader the metadata of all members (their subscriptions, client IDs, etc.).
• Sends all other members only their own membership info

The Group leader gets a list of all group members and their topic subscriptions from the coordinator. Based on the group’s configured strategy (Range, RoundRobin, Sticky, CooperativeSticky, etc.), it computes which partitions go to which consumer. It then send all assignements back to Coordinator which then distributes the assignment to all consumers, completing the rebalance.

Kafka Consumer Rebalancing and Partition Assignment

In a Kafka consumer group, rebalancing is the process by which the group redistributes partitions among its consumers whenever the group’s composition or topic metadata changes.

When Does Rebalancing Happen?

Rebalancing is triggered in the following scenarios:

• A new consumer joins the group.
• An existing consumer leaves or crashes.
• The number of partitions in a subscribed topic changes.
• A consumer changes its subscription (e.g., subscribes to a new topic).
• An admin manually triggers a group reset or partition reassignment.

During a rebalance, consumption temporarily pauses. Once the rebalance is complete, each partition is assigned to exactly one consumer in the group, ensuring load balancing and fault tolerance.

Who Coordinates the Rebalance?

The Group Coordinator, which is a broker selected per consumer group, is responsible for managing group membership and triggering rebalances.

Each consumer specifies its supported partition assignment strategies (e.g., Sticky, Range) in its configuration. When all consumers join the group, the Group Coordinator selects a common strategy shared by all and elects one Group Leader. The coordinator sends metadata about topics and consumers to this leader. The Group Leader then runs the selected strategy (e.g., StickyAssignor) to assign partitions to consumers and sends the result back to the coordinator. Finally, the coordinator distributes these assignments to all consumers in the group.

It’s generally recommended to use the StickyAssignor or CooperativeStickyAssignor to ensure minimal disruption during rebalances.

Static Membership and group.instance.id

By default, when a consumer disconnects (even briefly), the coordinator removes it from the group and triggers a rebalance. To avoid this, Kafka introduces static membership using the configuration parameter group.instance.id.

Each consumer in the group can be assigned a unique, static identity via group.instance.id. If a consumer disconnects and reconnects within the session timeout using the same ID, the coordinator recognizes it as the same instance—preventing unnecessary rebalancing. This is particularly useful in environments where transient network issues or restarts are common.

Handling Rebalance Events in Code

Kafka provides the ConsumerRebalanceListener interface to allow developers to manage offsets safely during rebalances. It includes two key callback methods:

• **onPartitionsRevoked(): Called before the consumer loses ownership of partitions. Use this to commit offsets synchronously**, ensuring no processed messages are lost.
• **onPartitionsAssigned(): Called after partitions are reassigned. Use this to initialize or reset state** related to the newly assigned partitions.

Conclusion

Kafka’s consumer model provides a powerful and flexible mechanism for building scalable, fault-tolerant data processing systems. By understanding how consumer groups, partition assignment, offsets, and rebalancing work together, developers can design systems that achieve the desired delivery semantics — from at-least-once to exactly-once. Effective offset management, use of cooperative or sticky rebalancing, and static membership can greatly reduce disruptions during rebalances. Tuning configurations such as max.poll.interval.ms, heartbeat.interval.ms, and enable.auto.commit helps balance reliability and performance. Ultimately, mastering the internals of Kafka consumers — including how group coordination and partition assignment strategies operate — is key to building robust, low-latency stream processing applications.