Consumer Groups and Collaboration

Consumer groups solve two problems: who is responsible for processing which partition, and where to resume after a change. The former is ownership coordination, and the latter is progress persistence. Only when both work together can multiple processes divide and recover a single logical consumption task.

This article discusses standard Kafka consumer groups, excluding share groups. Read Kafka Architecture first to understand partitions and offsets; see Message Semantics for guarantees on handling duplicates and external side effects.

Grouping Means Shared Tasks, Not Shared Read Positions

Assume a topic has six partitions, and the inventory group has three members:

Within the same group, each partition is assigned to one member; different groups read the same log independently. G1 and G2's progress are independent; one group does not wait for the other to finish reading before it can start.

Six partitions can be assigned to at most six group members holding partitions; a seventh member might remain idle. Increasing partitions improves sharding parallelism but also changes key mapping and state distribution. Within a partition, worker threads can execute concurrently, but ordered completion, state contention, and offset commits become the application's responsibility.

Coordinator, Membership, and Assignment

The group coordinator manages member, assignment, and progress-related protocols. Members prove they are still participating in the group through heartbeats and other interactions. The coordinator arranges new assignments after members leave, time out, or subscriptions change.

It is important to distinguish two levels here: how the broker reliably saves group state, and how consumers obtain partition assignments. Consumers do not execute a round of Raft among themselves to reach consensus on assignments; broadly calling consumer groups a "consensus algorithm" obscures the actual protocol roles.

Membership also serves to isolate old instances. A process may still be running after losing contact; you cannot assume the old process has stopped writing to external databases just because a new member has taken over. Kafka can constrain its own group protocol and commits, but external side effects still require transactions, idempotency, or business-level fencing.

Rebalancing: Safe Handoff Is More Important Than Even Distribution

Taking the handoff of P1 from C1 to C4 as an example, safe handoff must handle at least these states:

If C1 crashes before it can commit, C4 replaying the recent segment is normal recovery behavior. If C1 has already lost ownership, you cannot treat the offset after old task completion as still authorized for commit; similarly, you must not allow it to write infinitely to external storage.

The cost of rebalancing includes downtime, cache warming, state recovery, and replay, not just recalculating an assignment table.

Revoked and Lost Cannot Perform Identical Cleanup

During normal revocation, the application can stop dispatching, wait briefly for in-flight tasks, and then attempt to commit the completed watermark; waiting cannot indefinitely block the group protocol. Allowing replay after a commit failure is easier to maintain processing guarantees than committing positions that were not fully completed.

When partitions are lost, new members may start working, and old members should revoke local task eligibility and clean up state. At this point, onPartitionsLost should not be treated as a final commit opportunity while still owning the partition. See ConsumerRebalanceListener for the differences between the two callbacks.

Local eligibility checks only protect this process's dispatching and callbacks. If an old task has already issued a request to the database, setting a cancelled flag cannot retract that request. You need external storage to atomically check authorization generation upon actual write, or deduplicate by stable event IDs; if you only check ownership once before sending the request, takeover can still occur between the query and the write.

Eager, Cooperative, and the New Consumer Protocol

"Gradual handoff" and "which generation of group protocol to use" are two dimensions and should not be simply equated. According to the Kafka 4.1 documentation:

The cooperative sticky assignor in Classic and the Consumer protocol introduced by KIP-848 are not the same feature. See Consumer Rebalance Protocol for protocol support, migration constraints, and server-side assignors.

The goal of Sticky is to reduce unnecessary movement; the goal of balancing is to distribute load. Even partition counts being uniform does not mean workload is uniform: a hot partition may be busier than ten others. Simply changing the assignor will not automatically split hotspots.

Offsets Are Recovery Promises, Not Fetch Counts

The committed value usually represents the position to read next. After processing offset 100, commit 101. What the application truly needs to maintain is the contiguous watermark representing "all records prior to this point are completed."

Even if 103 is completed, you cannot commit 104: a crash at this point would cause 101 to be skipped. Only after 101 succeeds can the watermark pass through the completed 102 and 103, advancing once to 104.

If you simply take the maximum offset completed by all worker threads, the commit will skip holes. Maintaining a set of uncompleted tasks and a contiguous watermark per partition allows converting parallel execution into reliable recovery progress. Progress can be independent across partitions; you do not need to let one slow partition block commits for all partitions.

The Contiguous Part Is the Prefix of Delivered Tasks, Not Every Integer Offset

If a poll for a partition returns [100, 103, 107], skipped positions may come from compaction or transaction filtering. Waiting for completion notifications for 101 and 102 would hang forever; taking the maximum completed offset directly would skip real uncompleted tasks. The correct basis is the task queue registered in read order.

The following model accepts only one batch at a time, accepting the next batch only after the current one is fully completed, making it easier to clarify recovery conditions. next_position is the position after this read for the partition; it can only be committed when all delivered tasks are completed. token is an application-local unique batch eligibility identifier, associated with this assignment for the current partition; it is not the generation field in the Kafka protocol.

class BatchProgress:
    def __init__(self, offsets, next_position, token):
        if offsets != sorted(set(offsets)):
            raise ValueError("offsets must be strictly increasing")
        if next_position < 0 or (offsets and next_position <= offsets[-1]):
            raise ValueError("Read end position must be after the last record")
        self.offsets = list(offsets)
        self.next_position = next_position
        self.token = token
        self.active = True
        self.done = set()
        self.front = 0

    def completed(self, offset, token):
        if not self.active or token != self.token:
            return False  # Old assignment, old batch, or callback after revocation
        if offset not in self.offsets:
            raise ValueError("This batch did not dispatch this offset")
        self.done.add(offset)
        while (self.front < len(self.offsets)
               and self.offsets[self.front] in self.done):
            self.front += 1
        return True

    def committable(self):
        if not self.active:
            return None
        if self.front == len(self.offsets):
            return self.next_position
        return self.offsets[self.front]  # Position of the earliest uncompleted task

    def revoke(self):
        self.active = False

old = BatchProgress([100, 103, 107], 110, token="assignment-A/batch-1")
assert old.committable() == 100
assert old.completed(107, old.token)
assert old.committable() == 100
assert old.completed(100, old.token)
assert old.committable() == 103
assert old.completed(100, old.token)  # Duplicate completion notification does not advance twice
assert old.committable() == 103
old.revoke()
assert not old.completed(103, old.token)
assert old.committable() is None

new = BatchProgress([103, 107], 110, token="assignment-B/batch-1")
assert not new.completed(103, old.token)
assert new.committable() == 103
assert new.completed(107, new.token)
assert new.completed(103, new.token)
assert new.committable() == 110
assert BatchProgress([], 110, token="empty").committable() == 110

Here, committing 103 after 100 is completed is safe because this sequential read has already determined that 101 and 102 have no business records to dispatch, and 103 is not yet completed. After all are completed, committing 110 can also skip control records that have already passed the read position; you cannot skip offsets based on guesswork without reading them. Real clients need to capture the next offset after reading per partition and bind it to the corresponding batch, avoiding using the position from a later poll to replace old batch boundaries. See KafkaConsumer.position for position definitions.

In the example, it is assumed that 103's recovery position has been successfully committed before the new member takes over from 103. Although 107 was executed by the old member, it may still be replayed; progress tracking prevents omissions but does not eliminate duplicate effects. The model also does not call the Kafka commit interface: committable() is just a candidate value; whether the commit succeeds depends on the consumer thread owning the partition. Worker threads return results through the completion queue and do not directly call the same KafkaConsumer concurrently.

A multi-batch pipeline can extend the model to an ordered queue per partition, but it must retain the read boundary, assignment identity, and uncompleted tasks for each batch; you cannot directly overwrite the previous batch's state with each poll.

Why Can't We Commit 108 Directly After Completing 107?

The progress bar should follow the order of delivered tasks, not the largest completed number. Below, completing 107 first, then 100, will leave the commit position stuck at the still-uncompleted 103; only after all are completed will it jump to the known next position 110. Candidate watermarks and persistent watermarks are displayed separately; crashes only use the latter for recovery, so "already done" may still require replay.

Auto Commit and Manual Commit

Auto commit periodically saves the consumption position based on client mechanisms; it does not know whether business tasks in the application thread pool are completed. Synchronous fetching and processing, versus asynchronous dispatching immediately after fetching, are two different usage patterns.

Manual commit provides control but does not automatically eliminate duplicates: there is still a crash window between business completion and successful commit. Asynchronous commits also need to handle request completion order and failure callbacks; you cannot blindly retry old offsets after failure, overwriting newer progress.

When messages cannot be processed, you need a clear business strategy: block the partition and wait for repair, or persistently record the failure and transfer it to a dedicated channel. Simply skipping and committing essentially changes the promise that "all messages have been successfully processed." See KafkaConsumer Javadoc for specific APIs and thread restrictions.

Heartbeats, Processing Timeouts, and Backpressure

"Process is alive" and "business is progressing" are not the same thing. Heartbeats help determine member liveness; constraints like poll intervals are also used to discover applications that cannot participate in the consumption protocol in time. Only increasing timeouts delays misjudgment but also delays takeover after true failures.

Assume an input rate of 1000 records per second and a processing capacity of 800 records per second; the backlog grows by approximately 200 records per second. Over ten minutes, this is about 120,000 records. Simply increasing client buffers does not eliminate this difference; it only moves the backlog from the broker to process memory.

Backpressure should match fetch and processing capacity: limit the number of in-flight tasks, pause partition fetching if necessary, while continuing to participate normally in the group protocol. Batch size, number of worker threads, and poll frequency must be designed together. For partitions requiring order preservation, you can parallelize across partitions but serialize within partitions.

Pause Thresholds Must Leave Room for One Already-Fetched Batch

Assume the local allowed retention is 1200 tasks, accepting at most 200 at a time. If you wait until the queue reaches 1200 to pause, the batch already returned in the current poll might push it to 1400. You can pause when the total of in-flight and queued tasks reaches 1000, and resume when it drops below 400; the gap between the two thresholds avoids frequent pausing and resuming for every completed task.

This budget assumes capacity is checked before dispatching and there is only one batch of up to 200 items pending acceptance. If there are multiple prefetch batches, all must be included in the budget. Rate-limiting by count does not equal limiting memory by bytes: 200 large messages and 200 small messages have different memory costs.

Kafka's max.poll.records limits the number of records returned per poll, not the data already cached in the underlying fetch. Therefore, task limits must be designed together with byte budgets; during pauses, you must still call poll as required by the client to participate in coordination; you cannot block the consumer thread with the work queue until you lose membership. See Consumer Configs: max.poll.records and max.poll.interval.ms.

How to Interpret Lag

Common lag is the difference between a certain log boundary and the committed position. This number is suitable for observing trends but does not always equal the number of pending business records, nor can it be directly converted into latency: record size, holes after compression, and partition traffic may all differ.

At a minimum, observe simultaneously:

• Lag per partition and the age of the oldest unprocessed event, to avoid total volume masking single-partition hotspots.
• The distance between fetch position and commit position, to determine if the backlog is on the broker or within the application.
• Processing throughput, latency distribution, and in-flight task volume, to identify downstream bottlenecks.
• Rebalance frequency and partition recovery time, to distinguish between stable slow consumption and repeated takeovers.

These metrics describe "how much is left," "how long to wait," and "where it is stuck." They can be combined with service objectives in Metrics and Alerts.

Comparison with Queue Consumers

RabbitMQ's classic queues and quorum queues focus on message dispatch and acknowledgment, while Kafka's standard consumer groups focus on partition division of labor and replay for logs. However, RabbitMQ also has Streams, so you cannot generalize the entire product as "acknowledgment means delete, never replayable." See RabbitMQ Streams.

Selection should compare specific models: do you need independent acknowledgment per event or continuous position commits, do you need to retain history or release upon completion, do you need multiple groups for independent replay or multiple workers competing for tasks. Kafka provides intra-partition ordering, but not global ordering for the entire topic.

References

• Kafka: Consumer Rebalance Protocol
• KafkaConsumer: Threads, Commits, and Consumption Patterns
• RabbitMQ: Streams
• Message Semantics: Duplicate Execution and Business Effects

Keywords: consumer group, coordinator, rebalance, cooperative, offset, contiguous watermark, backpressure, lag, fencing

Verify backlog recovery separately from commit correctness

A correct commit boundary does not prevent growing backlog. Capacity and overload recovery relates catch-up time to arriving work, spare capacity, and retries; control in computing systems adds scaling delay and oscillation. Extra consumers improve effective throughput only within partition, storage, and processing constraints. They do not replace durable commit and deduplication protocols.