Consumer Groups and Rebalancing

A consumer group shares partition ownership. The application still decides when an effect is complete and which offset is safe to commit. Most serious consumer bugs occur where those two decisions are confused.

Scope: Apache Kafka 4.0.x Java consumer using ordinary consumer groups. The algorithms below are explicitly pseudocode, not a compiled service. Kafka’s share-group APIs have a different model and are outside this chapter.

Ownership, position and durable progress

A group coordinator manages membership and assignment. In a conventional group, one member owns a partition at a time, though a member can own several partitions. Extra consumers beyond the assignable partitions cannot increase active group parallelism. Application worker threads are a separate concurrency decision.

poll() advances the consumer’s position as records are returned. That position says nothing about whether a database write, HTTP call or computation finished. A committed offset is the next position from which the group should resume after reassignment/restart.

Checkpoint: record 12 finished—can we commit 13?

Given: one partition returns offsets 10, 11 and 12, and its next fetched position is 13. Workers apply independent, replay-safe effects. Offset 11 is slow; 10 and 12 finish first. No skip policy permits abandoning 11.

Predict the largest safe committed offset, then test it by crashing the process:

1. Try 13. Restart begins at 13. Record 11 is unfinished but will not be delivered again through normal group recovery. The candidate fails.
2. Try 12. Restart begins at 12. Record 11 is still skipped. This also fails.
3. Try 11. Restart delivers 11 again, followed by 12. Nothing unfinished is skipped; the repeated effect for 12 is safe under our assumption.
4. Let 11 finish. Now every delivered record below 13 has a durable effect. Committing 13 becomes safe.

The rule follows from restart behavior: a committed position must not pass the first unfinished delivered record. “Highest completed offset plus one” violates that rule whenever completion has a gap.

Change one assumption: if the effects must execute in partition order, allowing 12 to run before 11 was already wrong. Correct offset bookkeeping cannot repair reordered business effects; serialize that work too.

“Prefix” refers to delivered offset order, not consecutive integers. Compaction and transaction control records can create numerical gaps. Preserve offset/leader-epoch metadata where the client API exposes it; use the batch’s next-offset metadata after fully processing a partition’s returned records.

No-argument commitSync() commits fetched positions. It is appropriate only when those positions correspond to completed work. Calling it after each record in a partly processed batch does not make that a per-record commit. The KafkaConsumer API documents position, commits and threading.

A synchronous recipe that fails closed

For a simple sink, synchronous partition-order processing is often easier to defend than a worker pool. Disable auto-commit. Bound each effect’s duration and poll batch so processing plus commit overhead stays within the poll-liveness budget.

# Java consumer fragment; group/bootstrap/deserializers are application-specific.
enable.auto.commit=false
# Select this when inputs may contain transactions:
isolation.level=read_committed

Pseudocode:

poll records
for each assigned partition's returned records, in offset order:
    for each record:
        apply the durable effect using a replay-safe event identity
        only after success, record the next completed position
    when the partition batch is complete, retain its next-offset metadata
commit the explicit completed positions while ownership is valid
repeat

if processing fails:
    stop processing that partition; do not advance past the failed record
    commit only already completed prefixes if ownership still permits it
    stop/restart from committed progress, or explicitly seek to retry
    do not continue fetching as if an uncommitted record will rewind itself

If a database effect commits and the Kafka offset commit fails, the effect can run again after restart. Put event deduplication and the mutation in the same database transaction, or make the mutation intrinsically replay-safe. A separate “seen ID” cache is insufficient if it can disagree with the durable effect.

If storing offsets in the database instead, atomically store output and progress, restore from those offsets, and fence partition ownership. Merely disabling Kafka auto-commit does not provide that transaction or ownership fence.

Backpressure needs a bounded queue

Kafka consumers pull. Slowing consumption can protect a destination, but it does not automatically slow producers: the topic backlog grows. If processing remains slower than arrivals, finite retention eventually makes the backlog unrecoverable.

A worker pool can keep polling responsive while effects run. It also creates another queue, complicates progress commits and can reorder effects. Give it an explicit byte budget, not just a record count: record sizes and deserialized objects vary.

Owner thread: poll → bounded partition queues → workers → completions
                  ▲                              │          │
                  │                              └──────────┘
                  └── pause / resume       completed-prefix tracking
                                                    │
                                            explicit offset commit

Owner thread: poll → bounded partition queues → workers → completions
                  ▲                              │          │
                  │                              └──────────┘
                  └── pause / resume       completed-prefix tracking
                                                    │
                                            explicit offset commit

State-machine pseudocode:

one owner thread performs all consumer API operations
maintain an ownership epoch, ordered pending work and byte count per partition

on each owner-loop iteration:
    drain worker completion notices
    accept a notice only for the still-valid ownership epoch
    advance each completed prefix; never jump an unfinished record
    pause partitions whose queued/in-flight budget reaches the high watermark
    poll within max.poll.interval.ms, even when all partitions are paused
    account for all returned records before dispatching more work
    dispatch only within the worker and destination concurrency limits
    commit explicit completed prefixes while ownership is valid
    resume eligible partitions below the low watermark

on revocation:
    stop dispatch for revoked partitions
    drain only within a bounded rebalance budget
    commit completed prefixes while still allowed; invalidate that ownership epoch
on already-lost ownership:
    invalidate the epoch; do not try to commit as the old owner

Worker completion order and effect execution order are different. Tracking the completed prefix prevents skipped work, but it does not undo a later database mutation that raced ahead. If effects require partition order, allow one in-flight sequence per partition. If only per-key order is needed, use key-serial execution while still tracking the partition prefix.

Ignoring stale completion notices protects offset bookkeeping, not the remote system. A request already executing can complete after revocation. Use destination idempotency/version checks or fencing when that stale effect would be harmful. Pause state and queue ownership must be reconciled after reassignment.

Follow a partition handoff

The order service’s search projection owns partition P on consumer A. Its local worker token says ownership epoch 7; that is application bookkeeping, not a substitute for a broker or destination fence.

1. A dispatches E17 to the search store, but has not committed the corresponding Kafka progress.
2. A loses P. It invalidates epoch 7 and stops dispatch. The group subsequently assigns P to B, which resumes from committed progress.
3. The old remote request succeeds after A lost ownership. Ignoring its completion notice prevents a stale progress update; it does not undo the search mutation.
4. B replays E17. A destination dedupe/version contract must make that safe, or a destination-enforced fencing token must reject obsolete work where stale writes would be harmful.

For an orderly revocation, A may have a bounded opportunity to drain and commit while still authorized. After ownership is already lost, it must not assume that opportunity remains. On a whiteboard, keep three columns: group assignment, local pending work, and destination state. A rebalance changes the first without atomically clearing the other two.

A fetch limit is not a memory ceiling

max.poll.records limits how many records one poll returns; it does not limit underlying fetches. The client can cache fetched records across polls. It may fetch from multiple brokers, and an oversized first batch can exceed configured fetch byte limits so progress remains possible.

Budget memory for client fetch buffers, decompression, deserialized objects, queued work and in-flight effects. Leave capacity for a returned batch after the high watermark is crossed; do not discard already-fetched records because a worker queue is full. If the largest permitted batch cannot fit safely, lower accepted record/batch sizes or change the processing architecture.

fetch.min.bytes and fetch.max.wait.ms trade fetch efficiency against sparse-traffic waiting. A larger wait does not delay every fetch when sufficient bytes are already available. Measure destination freshness before increasing either setting. These details are specified in the consumer configuration reference.

Checkpoint: how much time does a bigger queue buy?

The destination stops accepting writes. The consumer is still receiving 3,000 records/s. Predict how long it can continue before it must pause.

Given: a simplified steady flow, 2,000 bytes of application memory per queued record and 12 MB of unused capacity below the pause watermark. Fetch buffers, in-flight work and space for an already-returned batch have separate reservations. Use decimal MB.

1. Find net accumulation: with zero completions, all 3,000 records/s stay queued.
2. Convert to memory growth: 3,000 × 2,000 = 6 MB/s.
3. Find time to the watermark: 12 MB / 6 MB/s = 2 seconds. Real fetch bursts can reach it sooner than this smooth-flow estimate.
4. Pause the affected partitions. Keep polling for group liveness. New source events now accumulate in Kafka rather than an ever-growing application queue; producers have not automatically slowed down.

Doubling the spare queue space buys four seconds, not a sustainable processing rate. The conserved quantity is unfinished work: it must remain somewhere, complete, or be rejected/dropped under an explicit policy.

Change one assumption: if the destination still completes 1,500 records/s, net growth is 1,500 records/s, or 3 MB/s. The original 12 MB lasts four seconds. Recovery still requires a later interval in which completions exceed arrivals.

Poll liveness and group protocol are separate

The application must poll within max.poll.interval.ms; background heartbeats do not make unlimited application stalls safe. Increasing that interval can accommodate legitimate work, but it also prolongs some failure detection. It is not a substitute for bounded effects.

Cooperative classic assignment reduces unnecessary revocation; it does not guarantee sub-second recovery. Static membership can reduce churn for short restarts with stable unique identities, but longer session tolerance delays reassignment after a real failure. Do not reuse one static identity across simultaneously active instances.

The official protocol guide and migration companion separate this client migration from the KRaft broker migration.

Errors, dead letters and shutdown

A poison record needs a product decision. Retrying indefinitely blocks the partition; skipping changes the stream’s semantics; moving to a retry topic can reorder later work. Make the retry budget, quarantine destination and replay owner explicit.

A dead-letter publication must be durable before committing past the failed input. If both are Kafka operations, a transaction can combine them. Without that atomicity, a crash can duplicate the dead-letter record; failed publication must not silently advance input progress. Keep source identity and failure context for replay.

For shutdown, stop admitting/dispatching new work, wake the owner thread, drain within a deadline, and commit only completed prefixes while ownership permits. Leave unfinished work uncommitted. A shutdown hook may call wakeup(); it should not race ordinary consumer calls from another thread.

Scale for recovery, not just arrival rate

Record lag is a useful signal, but it is not necessarily a count of business records: offsets can contain gaps. Fetched-position lag can look small while workers still hold unfinished effects. Track committed progress, oldest unfinished event age, worker queue bytes, effect latency, failures and time remaining before retention removes needed data.

For backlog Q, arrivals λ and effective recovery processing rate μ, drain time is Q / (μ − λ) when μ exceeds λ. Matching arrival rate merely stops backlog growth. Adding consumers helps only if assignable partitions, key distribution, broker reads and destination capacity allow more effective processing.

Before shipping a consumer, stop its destination, restart it mid-batch, fail one record while later work completes, and revoke a partition with requests in flight. The expected result is bounded memory and no progress committed beyond an unfinished required effect. The sizing chapter turns that recovery contract into capacity requirements.

Whiteboard: keep unfinished work recoverable

Guided variant. One poll returns offsets 50, 52, 55 in that order; numerical gaps are expected. Effects 50 and 55 complete, while 52 is pending. Fill the safe committed position __, then cross out the unsafe candidate 56 by tracing a restart. Assume effects may execute independently and repeat safely.

Independent problem. The search consumer receives 8,000 records/s but completes 5,000/s for six minutes. A queued record occupies 1,500 bytes; spare application space below its pause watermark is 18 MB, with fetch/in-flight reserves separate. Calculate time to pause and total system backlog growth. After recovery the destination completes 11,000/s while arrivals remain 8,000/s; calculate drain time. Finally, draw a crash after a DB mutation but before offset commit, followed by reassignment. Mark where each unfinished or repeated effect lives.

Pass criterion: account for every accepted record across memory, Kafka and destination state; never solve overload by silently discarding work or committing past it. The next chapter asks whether the backlog still exists when recovery needs it.