Kafka Consumer Rebalancing

A rebalance is not a hiccup the client library absorbs. Under the assignor Kafka still selects by default, every consumer in the group hands back every partition it owns and then waits — and what it waits for is the slowest member's return from your processing code.

The group coordinator is a broker that owns one partition of __consumer_offsets and, with it, the membership of your consumer group. It knows nothing about how fast you process records. It knows four states — Empty, PreparingRebalance, CompletingRebalance, Stable — and two clocks: the session timeout, driven by a background heartbeat thread, and the rebalance timeout, which for a consumer is max.poll.interval.ms.

That second identity is the whole lesson. The panel below runs ten minutes of a real group: one topic, twelve partitions, four consumers named c0–c3, and one scripted deploy that adds a fifth at 2:00 and removes it at 6:00. Every rebalance you see is caused by something in the model, and the log says which.

Leave everything at the defaults and look at 2:00. Two rebalances in ten minutes, a longest outage of about five seconds, and four tenths of one per cent of partition-time with nobody consuming. Each consumer went magenta for a moment: it handed back everything it owned before it was allowed to rejoin. That is a planned scale-out on a healthy group, and it is the version of a rebalance most people have in mind.

Now drag ms per record — c1 to 500 and change nothing else. The deploy is identical. The longest outage goes to about two and a half minutes, a third of all partition-time has no one consuming it, and the group loses close to half its throughput — because c0, c2 and c3 have already given their partitions back and the coordinator will not finish the join phase until c1 returns from the batch it is in the middle of. One consumer got slow. Nobody changed a config.

The rebalance timeout is max.poll.interval.ms wearing a different name

When a consumer sends its JoinGroup request it declares a rebalance timeout. The consumer client does not have a config called rebalance.timeout.ms — it sends max.poll.interval.ms. In GroupRebalanceConfig the line is literally rebalanceTimeoutMs = config.getInt(MAX_POLL_INTERVAL_MS_CONFIG) for the consumer protocol type. Kafka Connect, which has a genuine rebalance.timeout.ms, takes the other branch.

On the broker, the group's rebalance timeout is then the maximum over all its members. The comment in ClassicGroup says exactly that: the group's rebalance timeout in milliseconds. It is the max of all members' rebalance timeout. One consumer configured for a five-minute poll interval sets the deadline for everybody.

That deadline is how long the coordinator will sit in PreparingRebalance waiting for stragglers. When it expires, completeClassicGroupJoin removes every dynamic member that has not sent JoinGroup — the broker log line is Group X removed dynamic members who haven't joined — and proceeds with whoever is left. Under the eager protocol those remaining members have been holding zero partitions for the whole wait.

So the obvious remedy is a trap. Set ms per record — c1 to 700, so that its batch of 500 records needs 350 seconds and overruns the five-minute poll interval. Then drag max.poll.interval.ms to 15 minutes. The rebalance count drops from two to one — and every other number gets worse: partition-time not consumed goes from 53% to 59%, throughput lost from 70% to 80%. You removed a rebalance by telling the coordinator to wait longer for the member that was causing it, and it waited.

The control that actually helps is max.poll.records. Put max.poll.interval.ms back to five minutes, leave c1 at 700 ms, and pull max.poll.records down to 200: the longest outage falls from four minutes to about one, and throughput lost from 70% to 13%. At 50 records it is 30 seconds and 8%. Batch duration is max.poll.records × per-record cost, and it sets both how long c1 can overrun its deadline and how long the rest of the group waits for it to reach a poll boundary. Nothing about c1 got faster. The unit of work got smaller.

The rebalance a consumer inflicts on itself

Since Kafka 0.10.1 (KIP-62) liveness and progress are two different things, measured by two different threads. A background thread sends a heartbeat every heartbeat.interval.ms — 3 seconds by default — and the broker declares the member dead if it hears nothing for session.timeout.ms. That timeout has been 45 seconds since Kafka 3.0; it was 10 seconds through 2.8, which is why old runbooks tell you to raise it.

That interval is also how quickly a member finds out a rebalance has started, because it learns from the heartbeat response. Drag heartbeat.interval.ms from 3 seconds to 15 on an otherwise healthy group and the longest outage goes from about five seconds to about seventeen — nothing failed, the members simply took longer to notice they had been asked to rejoin. The documentation says as much: the value can be adjusted even lower to control the expected time for normal rebalances.

Progress is the other clock. The same background thread also checks whether the application thread has called poll() recently, and if it has not, it does something people rarely expect — it takes the consumer out of the group on purpose:

log.warn("consumer poll timeout has expired. This means the time between
    subsequent calls to poll() was longer than the configured
    max.poll.interval.ms ...");
maybeLeaveGroup(DEFAULT, "consumer poll timeout has expired.");

Read the shape of that. The consumer is alive. It is heartbeating. It is doing exactly the work you asked it to do. And it sends a LeaveGroup request, which puts the coordinator into PreparingRebalance immediately — no timeout has to expire, because a member volunteered to go.

The loop that follows is the storm. c1 exceeds the interval, leaves, and the group rebalances. c1's partitions move to c0, c2 and c3. c1 finishes its batch, discovers it is no longer a member, rejoins — and that rejoin is another rebalance. It gets partitions back, takes another oversized batch, and leaves again. Nothing breaks the cycle, because the thing driving it is the ratio between the batch and the deadline, and neither changes.

How fast the cycle turns is set by how long a batch runs, so build a fast one: untick deploy at 2:00 and 6:00 so nothing else can trigger anything, then set max.poll.records to 200, ms per record — c1 to 200, and max.poll.interval.ms down to 30 seconds. c1's batch takes 40 seconds against a 30-second deadline: it overruns by ten seconds, every time, forever. 26 rebalances in ten minutes, on a group nobody is deploying to, with a broker that is perfectly healthy.

Now pull max.poll.records to 100. The batch takes 20 seconds, the deadline is 30, and the count goes to zero. Not fewer — none. That is the shape of this failure: it is a threshold, not a gradient, and you are either under it or you are in a loop.

Everything about this looks like a broker problem from the outside. Consumer lag climbs on every partition, the group flaps between Stable and PreparingRebalance, and the broker log fills with rebalance notices. The broker is fine. One consumer's batch is too big for its own deadline, and it is the only thing wrong.

This is the same structural failure as a single straggler task in a Spark stage: a barrier that everyone must reach, and one member that reaches it late. Kafka's version is worse only because the barrier repeats.

The cooperative assignor changes the price, not the trigger

Leave the 26-rebalance storm running and switch the assignor to CooperativeStickyAssignor. The count goes up, to 35, because the cooperative protocol needs two rebalances wherever a partition changes hands. Throughput lost falls from 3.7% to 2.7%. The poll timeout neither knows nor cares which assignor you chose; all the assignor decides is what each rebalance costs.

The whole difference is one branch in ConsumerCoordinator.onJoinPrepare:

• Eager — revoke all partitions. The member calls onPartitionsRevoked for everything it owns and sets its assignment to the empty set before sending JoinGroup. From that moment it has nothing to consume.
• Cooperative — only revoke those partitions that are not in the subscription anymore, which in the normal case is none. The member joins holding everything, and because the consumer's poll loop calls updateAssignmentMetadataIfNeeded(timer, false) — note the false, meaning do not block waiting for the join — it keeps returning records for its own partitions while the rebalance runs.

The revocation still has to happen, just later and only where it is needed. After SyncGroup, in onJoinComplete, a cooperative member computes owned − assigned, revokes exactly that, and calls requestRejoin(). That is the second rebalance, and it is not a defect: the first rebalance is the synchronisation barrier that guarantees no partition has two owners. Watch the log with the cooperative assignor selected and you will see the pair, a few hundred milliseconds apart.

The assignor cooperates by refusing to hand out contested partitions. CooperativeStickyAssignor.adjustAssignment removes every partition that is moving from the new owner's assignment for that round, so between the two rebalances those partitions belong to nobody. Move the inspect the group at slider into that gap and the bottom row of the grid shows them. Cooperative rebalancing does not eliminate unowned partitions; it reduces them from all twelve to the two or three that actually had to move.

Where it does pay, it pays enormously, and the size of the payment is the length of the slowest member's batch. Turn the deploy back on, set ms per record — c1 to 500, and compare: eager loses 46% of throughput with a third of all partition-time unconsumed, cooperative loses 12% with under 4%. On a healthy group with four-second batches the same comparison is a wash — 0.5% against 0.9%, and cooperative is slightly worse, because two rounds take longer than one. Cooperative rebalancing is not free; it is insurance against a member that takes a long time to reach a poll boundary.

And it is insurance, not immunity. The coordinator still waits for c1's JoinGroup either way, so a newly deployed consumer gets nothing until c1 surfaces. What cooperative rebalancing changes is who waits: "the group stops" becomes "the new consumer waits", which is a large win and not the same as no wait.

One last thing, and it is the one that catches most teams: you are almost certainly not using any of this. The default value of partition.assignment.strategy in Kafka 4.1 is [RangeAssignor, CooperativeStickyAssignor] — a list ordered by preference, and the group picks the first strategy every member supports. RangeAssignor is eager. That list has been the default since 3.0 precisely so that you can drop RangeAssignor from it in a single rolling bounce, but until you do, nothing changes. Check what your group actually negotiated rather than assuming.

Static membership delays the rebalance, it does not prevent it

Setting group.instance.id makes a consumer a static member (KIP-345, Kafka 2.3). A static member keeps its member id across restarts, so a pod that comes back within session.timeout.ms reclaims its previous assignment and the group never rebalances. For rolling restarts this is the single most effective setting available.

Against a slow poll loop it is worse than nothing. Tick the group.instance.id set box with c1 at 700 ms and read the log. maybeLeaveGroup only sends the request when isDynamicMember(), so the static member says nothing — it simply stops heartbeating. The coordinator now waits the full session.timeout.ms before removing it, and for that entire window c1's partitions are assigned to a member that is not consuming them. The config documentation states the behaviour plainly: partitions will not be immediately reassigned. Instead, the consumer will stop sending heartbeats and partitions will be reassigned after expiration of the session timeout.

Put the 26-rebalance storm back — deploy off, max.poll.records 200, c1 at 200 ms, max.poll.interval.ms 30 seconds — and tick the box. The rebalance count halves, to 13, which looks like a win until you read the line beside it: partition-time not consumed goes from 4.9% to 9.0%. You bought half the rebalances by paying for them in dead partitions. Static membership is a deployment tool. It is not a fix for a consumer that cannot finish a batch.

The one control that reliably lowers the frequency of everything on this page is the size of the unit of work: max.poll.records, or handing the batch to a worker pool and calling poll() on schedule regardless. If you do the latter, you own the pause/resume and offset commit logic yourself, and getting it wrong trades rebalances for duplicates.

Checking it in a running system

Four consumer metrics, all in the consumer-coordinator-metrics group, settle this in about a minute:

• rebalance-rate-per-hour — on a stable group this is approximately zero. Anything above a handful per hour on a group nobody is deploying to means something is leaving on its own.
• failed-rebalance-rate-per-hour — non-zero means rebalances are being superseded before they complete, which is the storm's signature.
• rebalance-latency-max — compare this to max.poll.interval.ms. If it is close, the coordinator is waiting out the deadline for a member that never joined.
• last-rebalance-seconds-ago — the cheapest alert you can write. Graph it per consumer; a sawtooth is a storm.

Then read the logs, because between them they name the culprit outright. On the client, the warning quoted above appears verbatim, and it already tells you to lower max.poll.records. On the broker there are two lines, and you need both. The first identifies who left and why the client says it left — the reason string travels in the LeaveGroup request itself:

[Group orders] Member consumer-orders-3-a1b2 has left group through
  explicit `LeaveGroup` request; client reason: consumer poll timeout has expired.
Preparing to rebalance group orders in state PreparingRebalance with old
  generation 41 (reason: explicit `LeaveGroup` request for (consumer-orders-3-a1b2) members.).

The second line's reason only says that somebody left; the first says why, in the consumer's own words. Grep for client reason across your brokers and you have a census of every self-inflicted rebalance in the cluster. Reasons you will see there that are not self-inflicted look different — Adding new member ... with group instance id ..., removing members on heartbeat expiration — and that distinction is the whole diagnosis.

Finally, kafka-consumer-groups.sh --describe --group g --state prints the coordinator's view. A group that is repeatedly PreparingRebalance when you run it twice in a row is not recovering from a deploy. Add --members --verbose and check whether the member count is stable and whether the assignment moves between runs; a sticky assignment that keeps changing means partitions are circulating rather than being processed.

Next: the protocol that removes the barrier rather than widening it — the KIP-848 consumer group protocol, available as group.protocol=consumer in Kafka 4.x, which moves assignment to the broker and reconciles members one at a time instead of stopping the group at a global synchronisation point. The default is still classic in 4.1, so everything on this page still applies to most groups running today. Underneath both sits the group coordinator itself.