Kafka Consumer Rebalancing: Cooperative vs Eager Protocols Compared

How Kafka consumer rebalancing works

A consumer group spreads a topic's partitions across its members so that each partition is read by exactly one consumer at a time. Rebalancing is the process that recomputes that mapping whenever group membership or the set of subscribed partitions changes.

The coordinator, the leader and the two phases

Each group is managed by a group coordinator, a broker chosen by hashing the group ID. In the classic protocol, the coordinator does not compute assignments. It delegates that work to a client:

CopyMember A (leader)      Member B           Coordinator
      |                    |                    |
      |---- JoinGroup ---------------------->   |
      |                    |---- JoinGroup -->  |
      |                    |                    |  (waits for members,
      |<--- JoinResponse (all subscriptions) ---|   bumps generation)
      |                    |<-- JoinResponse ---|
      |  run assignor      |                    |
      |---- SyncGroup (assignments) --------->  |
      |                    |---- SyncGroup ---> |
      |<--- SyncResponse (A's partitions) ------|
      |                    |<-- SyncResponse ---|

During the join phase, every member sends its subscription. The coordinator increments the group's generation and picks a leader. The leader receives the full member list and runs the configured assignor. In the sync phase, the leader submits the assignments. The other members send empty requests, and the coordinator returns each member its share. Keeping the assignor on the client lets teams change assignment logic without upgrading brokers.

Triggers

A rebalance starts when:

• a member joins or leaves cleanly,
• heartbeats stop for longer than session.timeout.ms,
• the gap between poll() calls exceeds max.poll.interval.ms,
• a subscription changes, including a regex that starts matching a new topic,
• a subscribed topic gains partitions.

Newer Kafka releases add a broker-side consumer group protocol. It moves assignment to the coordinator and replaces the join/sync round trips. Support depends on both your client and your broker versions, so check their documentation rather than assuming it is available.

Copy+------------------------------+
| Rebalance triggers           |
|  - member joins or leaves    |
|  - session.timeout.ms expiry |
|  - max.poll.interval.ms gap  |
|  - subscription change       |
|  - partition count change    |
+------------------------------+
               |
               v
+---------------------------------------------------------------------+
| Group coordinator (broker chosen by hashing group.id)               |
| JOIN: collect subscriptions, bump generation, pick leader           |
| SYNC: store leader's plan, return each member its share             |
| Never computes assignments itself (classic protocol)                |
+---------------------------------------------------------------------+
      | JoinResponse      ^ SyncGroup      ^ JoinGroup +       | SyncResponse
      | (all subs)        | (assignments)  | empty Sync        | (own share)
      v                   |                |                   v
+--------------------------+           +------------------------------+
| Member A (group leader)  |           | Members B, C, ...            |
| sends JoinGroup, then    |           | wait for coordinator reply   |
| runs configured assignor |           | receive assigned partitions  |
+--------------------------+           +------------------------------+

In the classic consumer group protocol, a trigger starts a rebalance: the coordinator collects JoinGroup requests and bumps the generation, the elected leader runs the assignor and returns the plan in SyncGroup, and the coordinator gives each member its partitions.

Eager vs cooperative protocols: what actually differs

Under the classic protocols, the two approaches differ in what a member gives up when the group changes.

With the eager protocol, every member revokes all of its partitions before rejoining. If one partition needs to move because a single pod restarted, every consumer in the group still stops, commits, and waits for a new assignment. The pause scales with group size, not with how much actually changed.

Cooperative incremental rebalancing revokes only the partitions that are moving. Members keep consuming everything else during the first round. Freed partitions are handed out in a follow-up rebalance. Full balance can take two rounds, but most partitions never pause.

The assignor determines how much moves. Range and round-robin compute each assignment from scratch and ignore who owned what before. The cooperative sticky assignor balances the load and then keeps as many existing partition-to-consumer mappings as it can. This matters most for consumers that hold per-partition state: local state stores, in-memory caches, open connections, or deduplication windows. When a partition moves, that state has to be rebuilt elsewhere, and rebuilding it often costs more than the rebalance pause.

Why rebalance storms happen in containerized microservices

Platform teams feel this first. Kubernetes restarts consumers far more often than the group protocol expects, and each restart triggers a membership change. Every member leaving or joining starts a rebalance. When several of these overlap, the group spends more time reassigning partitions than consuming them.

Neither protocol fixes the poll-interval cause. Cooperative assignment makes each rebalance cheaper, but a consumer that misses its poll deadline still gets evicted. Raising max.poll.interval.ms only hides the slow consumer and delays detection of a stuck one. The real fix is to bound the work done per poll, either by lowering max.poll.records or by putting a time budget on each batch. The alternative is to move processing off the poll thread. If you do that, pause the partitions while work is in flight and commit offsets only after the work completes.

Implementing cooperative sticky assignment and static group membership

The configuration example sets four properties. partition.assignment.strategy points at CooperativeStickyAssignor, so the consumer revokes only the partitions that are moving and keeps the rest. group.instance.id gives each consumer a stable identity. session.timeout.ms is raised well above the default to cover a normal pod restart. max.poll.interval.ms is sized to your worst-case batch processing time.

In Kubernetes, use a StatefulSet. Pod names such as orders-consumer-0 and orders-consumer-1 stay the same across restarts, so you can inject metadata.name through the Downward API as an environment variable and use it as group.instance.id. A Deployment produces random pod suffixes, which gives every restart a new identity and defeats static membership.

The rebalance listener

In cooperative mode, onPartitionsRevoked receives only the partitions leaving this consumer. Commit offsets and flush local state for that subset only. Do not clear caches or reset processors globally, because the other partitions are still assigned and still processing. onPartitionsAssigned may be called with an empty collection, so make it safe to run as a no-op.

Handle onPartitionsLost separately. By the time it fires, another member may already own those partitions. Committing offsets there could fail or overwrite a newer position, so discard in-flight work and drop local state instead.

Session timeout on restart

A static member does not send a leave request when it shuts down. The coordinator waits for session.timeout.ms before acting. If the pod returns within that window with the same instance id, it gets its previous assignment back and no rebalance occurs. The cost is that a pod that has actually died leaves its partitions unconsumed for the full timeout. Set the value just above your observed restart time, not arbitrarily high.

Copyimport java.util.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

// Per-partition local state; flush() persists it, discard() drops it. App-defined.
interface PartitionState { void flush(); void discard(); }

class CooperativeListener implements ConsumerRebalanceListener {
  private final Consumer<String, String> consumer;
  // Next offset to commit per partition, updated by the poll loop after processing.
  final Map<TopicPartition, OffsetAndMetadata> pending = new HashMap<>();
  final Map<TopicPartition, PartitionState> state = new HashMap<>();

  CooperativeListener(Consumer<String, String> consumer) { this.consumer = consumer; }

  @Override public void onPartitionsRevoked(Collection<TopicPartition> revoked) {
    // Only the partitions leaving this consumer. The rest are still assigned: touch nothing else.
    Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
    for (TopicPartition tp : revoked) {
      PartitionState s = state.remove(tp);
      if (s != null) s.flush();
      OffsetAndMetadata o = pending.remove(tp);
      if (o != null) toCommit.put(tp, o);
    }
    if (!toCommit.isEmpty()) consumer.commitSync(toCommit); // commit the revoked subset only
  }

  @Override public void onPartitionsAssigned(Collection<TopicPartition> assigned) {
    // Often empty in cooperative mode; the loop must be a harmless no-op then.
    for (TopicPartition tp : assigned) state.computeIfAbsent(tp, k -> newState(k));
  }

  @Override public void onPartitionsLost(Collection<TopicPartition> lost) {
    // Another member may already own these: never commit here, just drop work and state.
    for (TopicPartition tp : lost) {
      pending.remove(tp);
      PartitionState s = state.remove(tp);
      if (s != null) s.discard();
    }
  }

  PartitionState newState(TopicPartition tp) { throw new UnsupportedOperationException("app-specific"); }
}

A ConsumerRebalanceListener for cooperative mode that flushes state and commits offsets only for the revoked partitions, tolerates empty assignments, and discards lost partitions without committing.

Migrating safely from eager to cooperative

Step one: advertise both assignors

When a consumer joins, it sends the list of assignors it supports. The group coordinator picks one that every member supports. If you flip one pod straight to cooperative-sticky only, while the rest of the group still lists only an eager assignor such as range, the members share no protocol. The new pod's join is rejected and it cannot consume. A fast rolling deploy can leave a mix of rejected and old members for the whole rollout.

To avoid this, deploy first with both assignors configured, cooperative-sticky listed first and your existing eager assignor second. Every member now has a protocol in common. The group still rebalances eagerly during this phase, because the client uses the eager protocol whenever any configured assignor needs it. What this bounce accomplishes is that every member advertises cooperative-sticky.

Step two: remove the eager assignor

After the first rollout has fully completed, do a second rolling bounce with only cooperative-sticky configured. From then on, rebalances are incremental.

Pre-migration checklist for listener code

• Revoke handler: confirm it does not assume revocation means "everything." It will receive only the partitions being moved. Clearing all local state or caches there will discard data for partitions you keep.
• Assign handler: confirm it does not assume it receives the full assignment. It now receives only the newly added partitions, so code that rebuilds state from that set will lose track of retained ones.
• Commits: make sure offsets for revoked partitions are committed in the revoke handler. Don't rely on a full-revocation commit that no longer happens.
• Lost partitions: handle lost partitions separately, without committing.

Kafka Streams and similar frameworks configure their own assignors and membership, so follow their upgrade guidance instead of this procedure.

Copy# consumer.properties: two-step rolling migration from eager to cooperative

# --- Rollout 1: advertise both assignors ---
# Cooperative-sticky is listed first so it is preferred once every member supports it.
# RangeAssignor stays in the list so upgraded pods share a protocol with pods
# that have not been bounced yet, and the coordinator accepts their joins.
# Rebalances remain EAGER during this phase because an eager assignor is still configured.
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor,org.apache.kafka.clients.consumer.RangeAssignor

# --- Rollout 2: run only after rollout 1 has finished on every pod ---
# Drop the eager assignor. Every member already advertises cooperative-sticky,
# so the group keeps a common protocol, and rebalances become incremental.
# partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# Leave group.id unchanged across both rollouts so members stay in the same group.
group.id=orders-processor
bootstrap.servers=kafka-1.example.internal:9092,kafka-2.example.internal:9092

Consumer configuration for the two rolling bounces: the first lists cooperative-sticky ahead of the existing range assignor, and the second, applied only after the first completes, removes range so rebalances become cooperative.

When to avoid it, failure modes and alternatives

None of these options is free. Each one removes a specific kind of disruption and adds a different failure mode.

For small, stateless groups that rarely change size, eager rebalancing is often good enough. A brief stop-the-world pause costs less than the extra operational complexity. Where your brokers and clients support the broker-side consumer group protocol, assignment moves to the coordinator and members reconcile incrementally, which avoids much of the client-side coordination described above. If you run a stream processing framework, let it manage membership. Overriding its assignor usually breaks its assumptions about state placement.

Every one of these techniques reduces duplicate processing but does not eliminate it. A rebalance can still occur between processing a record and committing its offset. Your consumers still need idempotent writes or transactional processing.

Observing rebalances and the impact on cost and engineering time

Rebalances are visible from three places. Use all three, because each one covers a gap in the others.

• Broker view. kafka-consumer-groups --describe --group <name> shows current assignments and per-partition lag. Adding --members or --state shows member count and whether the group is Stable or PreparingRebalance. This is a point-in-time snapshot, so it is good for live debugging but not for trends.
• Client metrics. The Java consumer exposes coordinator metrics such as rebalance-total, rebalance-latency-avg, rebalance-latency-max, and failed-rebalance-total. Graph the rate of rebalance-total, not the raw counter. A rebalance rate that stays elevated outside deploy windows usually indicates a storm.
• Lag. Consumer lag (for example, records-lag-max) shows the cost of a rebalance. Under eager assignment, lag spikes across all partitions. Under cooperative assignment, lag should rise only on the partitions that moved.

To correlate these signals, annotate your dashboards with deploy events and pod restart or OOM-kill events from your orchestrator. If a rebalance spike lines up with a restart and not with a deploy, look at liveness probes and max.poll.interval.ms before you look at the assignor.

Illustrative estimate (assumed numbers, not a benchmark).

Monthly savings = (incidents avoided × hours per investigation) + (cleanups avoided × hours per cleanup)

Assume the migration takes 3 incidents and 2 duplicate-data cleanups off your monthly workload. Assume each investigation takes 4 hours and each cleanup takes 6 hours. That gives 3×4 + 2×6 = 24 engineer-hours per month.

Now assume the one-time cost is 40 hours: two coordinated deploys, an audit of rebalance listeners, and new alerts. With these numbers, the migration pays back in about two months. If your team rarely sees rebalance incidents, plug in your own numbers. The payback period can be much longer.

Copy#!/usr/bin/env bash
# Poll a consumer group's state on a fixed interval and log each snapshot with a UTC timestamp.
# This turns the point-in-time --describe output into a timeline you can line up
# against deploy and pod-restart events.
set -euo pipefail

BOOTSTRAP="kafka-0.kafka.internal:9092"   # placeholder broker address
GROUP="orders-consumer"                    # placeholder consumer group
NAMESPACE="orders"                          # placeholder Kubernetes namespace
INTERVAL=15                                  # seconds between snapshots
CMD="kafka-consumer-groups.sh"              # use kafka-consumer-groups on some distributions

while true; do
  ts=$(date -u +%Y-%m-%dT%H:%M:%SZ)

  # Group state (Stable, PreparingRebalance, CompletingRebalance), member count,
  # and assignment strategy, which confirms which assignor is active.
  "$CMD" --bootstrap-server "$BOOTSTRAP" --describe --group "$GROUP" --state \
    | sed "s/^/$ts state   /"

  # Per-partition lag. Under eager assignment, lag rises on every partition at once.
  # Under cooperative assignment, it rises only on the partitions that moved.
  "$CMD" --bootstrap-server "$BOOTSTRAP" --describe --group "$GROUP" \
    | sed "s/^/$ts lag     /"

  # Pod restart counts. If these increase at the same moment the group leaves Stable,
  # check liveness probes and max.poll.interval.ms before you change the assignor.
  kubectl get pods -n "$NAMESPACE" --no-headers \
    -o custom-columns=NAME:.metadata.name,RESTARTS:.status.containerStatuses[0].restartCount \
    | sed "s/^/$ts pods    /"

  sleep "$INTERVAL"
done >> "rebalance-timeline-${GROUP}.log"

A polling script that records consumer group state, per-partition lag, and pod restart counts with UTC timestamps in one log, so you can match rebalances to deploys and restarts.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to kafka consumer rebalancing, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Reviewers see the exact diff and the list of affected services up front, so approval is a decision rather than an investigation.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• Per-service approve/reject means a bad shared value can be refused where it would break something, before it is live.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.