Kafka Rebalance Deep Dive: Eager vs Cooperative Protocols and Eliminating Partition Transfer Pauses

Problem Scenario: Rolling Deploy Causes Full Consumption Pause

A consumer group with 12 partitions and 4 consumers experiences a lag spike every time an instance restarts during a rolling deploy. All 12 partitions stop consuming simultaneously for tens of seconds. Even scaling out — moving only 3 partitions to a new instance — pauses the other 9 partitions. The pause is not about how fast the assignment algorithm runs; the classic protocol requires members to surrender all partitions before a new assignment is computed, so even partitions that return to the same consumer are interrupted.

What Triggers a Rebalance

A rebalance redistributes partitions among consumers and increments the generation number, invalidating the previous assignment. Triggers include:

Member changes: new instance calls subscribe() to join; instance calls close() to leave gracefully (active changes). Process crashes and network partitions are passive changes detected via heartbeat timeout.

Subscription changes: topic list changes or partition count increases.

Coordinator change: the group coordinator is the leader of a __consumer_offsets partition. When the coordinator moves, the whole group reloads and all members must rejoin — a detail often overlooked.

Classic Protocol Flow

The classic protocol has three phases:

JoinGroup : each member reports subscribed topics; coordinator waits for all, elects a leader.

SyncGroup : leader computes assignment with the assignor, sends to coordinator, which distributes to each member.

Stable period : members maintain membership via heartbeats.

The barrier that requires all members to check in means the slowest member determines the rebalance duration. During this barrier, even members that still hold partitions cannot continue consuming.

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-fulfillment");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
// Process then commit to avoid offsets running ahead of business
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> parts) {
        // Partition about to leave, commit processed offsets first
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> parts) {
        log.info("assigned {}", parts);
    }
});

In the eager protocol, onPartitionsRevoked fires for every partition on every rebalance, including those that will immediately return. This callback runs inside

poll()</sub>; its execution time directly adds to the group pause. It should only commit offsets — nothing else.</p>
<h2>Eager Protocol: Global Pause</h2>
<p>Eager means "release all, then re-claim". Upon rebalance notification, a member calls <code>onPartitionsRevoked

for every partition it holds, then sends JoinGroup to rejoin. Until the new assignment arrives and the consumer fetches data, no partition has an owner. Offsets stall; lag climbs steeply.

Pause duration stacks up:

Coordinator detects join or heartbeat timeout, sets state to PreparingRebalance, waits for all members to JoinGroup.

Member processing thread must exit poll(). If it's processing a batch of 500 records, this step waits for that batch to finish — often the slowest segment. Every extra second in business code adds a second to the whole group pause.

Leader computes assignment, SyncGroup distributes, consumers issue Fetch requests; first data batch marks recovery.

Coordinator won't wait forever; stragglers are treated as left, wasting the round.

A rolling restart of N instances theoretically triggers N rebalances, but often more: new instance start triggers one, old instance close() triggers another. If close together, the coordinator may merge them or run two separate pauses, producing consecutive lag spikes. group.initial.rebalance.delay.ms (default 3000 ms) makes a new group wait before first assignment to batch simultaneously starting members. Don't set to 0 in production or a wave of simultaneous starts will trigger dozens of rebalances.

Cooperative Protocol: Partition Retention

KIP-429 (Kafka 2.4) introduced the incremental CooperativeStickyAssignor. On join, members include ownedPartitions in their JoinGroup metadata — the fundamental difference from eager.

The leader still computes a target assignment using sticky logic, then adds a filtering step: if a partition is assigned to A in the target but currently held by B, A does not receive it yet; only B is told to revoke. A starts with the reduced assignment; the gap is filled in the next round. A second round exists because the leader cannot assign a partition to a new owner before the old owner releases it, otherwise two consumers would process the same data and overwrite offsets. After B revokes and rejoins, a second rebalance occurs. The group repeats rounds until a round has no revocations; only then does the partition truly move to A.

Benefit: unchanged partitions keep consuming. Scaling from 12 partitions to 5 consumers interrupts only 2–3 partitions; the other 9 continue fetching throughout. Round count increases from one to two, total time may be slightly longer than eager, but interrupted partitions drop by an order of magnitude — a worthwhile trade-off.

Cost: callback semantics change. Under cooperative, onPartitionsRevoked receives only partitions actually being revoked; code cannot assume it gets all partitions or perform global cleanup. Kafka 2.4 added onPartitionsLost for when a member is deemed dead and its partitions already reassigned — offset commits will fail there. Cleanup logic must split: safe actions in revoked, forced actions in lost. Mistakes cause duplicate processing or orphaned temporary files.

Upgrade Path

Since Kafka 3.0, partition.assignment.strategy defaults to [RangeAssignor, CooperativeStickyAssignor]. The group picks a protocol all members support, so default groups still run eager Range. To switch to cooperative, remove RangeAssignor from the list and do a rolling restart. If members only list eager assignors, two rolling restarts are needed.

// Phase 1: both assignors present, group still runs eager
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        List.of(CooperativeStickyAssignor.class, RangeAssignor.class));

// Phase 2: after all instances upgraded, keep only incremental
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        List.of(CooperativeStickyAssignor.class));

Assignor Comparison

RangeAssignor : eager, revokes all. Default; maps topic partitions 1:1.

RoundRobinAssignor : eager, revokes all. Better distribution for multi-topic subscriptions.

StickyAssignor : eager, revokes all. Name suggests stickiness, but only keeps assignment result similar; members still surrender all partitions. Scaling still pauses the whole group — it saves assignment result, not the pause.

CooperativeStickyAssignor : cooperative, revokes only moving partitions. For stateful consumers, frequent deploys.

Heartbeat, Poll Interval, and Member Identity

Classic protocol uses two independent timeouts: session.timeout.ms (default 45000): checks if heartbeats arrive on time. Heartbeats sent by a background thread, independent of business thread; detects process crashes, network partitions. max.poll.interval.ms (default 300000): checks interval between two poll() calls. Detects slow business processing.

They trigger different paths: heartbeat timeout — server evicts member, partitions become unassigned. Poll timeout — client self-detects overload, sends LeaveGroup, then partitions are revoked. Logs show consumer poll timeout has expired before the rebalance.

Default math: max.poll.records=500, 30 ms per record = 15 seconds per batch, far from 300 seconds. Real danger: occasional slow SQL or downstream timeout drags a batch to 5–6 minutes. No poll() calls during that time; client thinks it's stuck, sends LeaveGroup. Batch finishes but offsets uncommitted; partitions already reassigned — duplicate processing. Many unexplained duplicate consumptions originate here.

Parameter Reference

session.timeout.ms

(45000): heartbeat arrival; member removed from group. heartbeat.interval.ms (3000): heartbeat frequency; recommended 1/3 of session timeout. max.poll.interval.ms (300000): poll interval; client leaves group. max.poll.records (500): records per batch; determines batch processing time. group.initial.rebalance.delay.ms (3000): new group first assignment wait; affects startup latency only.

Stuck in PreparingRebalance

Group stays in PreparingRebalance with all members present and heartbeats healthy. Common cause: business thread blocked by long GC or synchronous remote call, poll() doesn't return. Coordinator's rejoin signal unprocessed. From coordinator's view, heartbeats arrive on time — no reason to evict — so it waits until a timeout forces a reshuffle. Monitoring shows abnormal rebalance duration; root cause is JVM or downstream dependency.

Static Membership with group.instance.id

group.instance.id

gives a consumer a fixed identity, replacing random member.id on each restart. If the same id rejoins within session.timeout.ms, coordinator returns its original partitions — no rebalance. For services deploying multiple times a day, this eliminates most planned pauses. Trade-off: if a static member truly crashes, its partitions wait for session.timeout.ms to expire before reassignment. Raising timeout from 45s to 120s makes restarts smoother, but a real crash adds two minutes of zero consumption for that partition. Both directions can't be optimized simultaneously; choose based on tolerance for downtime vs duplicate processing.

Hard constraint: two running instances cannot share the same group.instance.id. The later one gets fenced with FencedInstanceIdException and shuts down. Kubernetes rolling update (new pod up before old pod down) with instance id derived from StatefulSet ordinal causes exactly this conflict during deploy windows.

KIP-848: Server-Side Assignment (Kafka 4.0)

Kafka 4.0 graduates the new consumer group protocol (KIP-848) with group.protocol=consumer. Assignment moves to broker; coordinator maintains member-partition mapping. Member joins/leaves affect only their partitions; others receive incremental notifications. Changes are per-partition, no global barrier.

Several client parameters become unsupported in this protocol: session.timeout.ms, heartbeat.interval.ms, partition.assignment.strategy — setting them throws ConfigException. Equivalents move to broker config: group.consumer.session.timeout.ms (default 45000) group.consumer.heartbeat.interval.ms (default 5000)

Assignors via group.consumer.assignors (default uniform,range); client picks one via group.remote.assignor from broker-provided list. max.poll.interval.ms and group.instance.id remain effective on both sides; former still client-enforced, latter still static membership credential.

Migration is gradual. group.consumer.migration.policy defaults to bidirectional: classic and new-protocol members coexist. First new-protocol member triggers group conversion; coordinator translates classic JoinGroup/SyncGroup/Heartbeat into new protocol calls until no classic members remain.

kafka-consumer-groups.sh --describe --group order-fulfillment --members --verbose

shows an UPGRADED column indicating which instances have switched.

Pause Composition and Troubleshooting Order

Back to the deploy scenario. Lag spike height = number of partitions simultaneously without owner (eager = all, cooperative = only moving). Width = time from coordinator entering PreparingRebalance to each consumer rejoining, leader computing assignment, issuing Fetch, and receiving first data batch. Former is protocol-determined; latter depends on member responsiveness: how fast processing thread exits poll(), what onPartitionsRevoked does, coordinator wait for new members.

If a deploy shows lag spike but rebalance count unchanged, the problem isn't the protocol — look at per-batch processing time and downstream response latency; those explain why offsets stall.

Both segments are measurable; only two knobs are directly adjustable: time spent in poll() and actions in onPartitionsRevoked. To narrow width:

Check onPartitionsRevoked for anything beyond commitSync. This callback runs synchronously on the poll thread; a DB write or RPC call multiplies by member count into total duration.

Reduce max.poll.records and per-record processing time. If a batch can't finish, the thread can't respond to rejoin notification. Dropping batch size to 50–100 is usually more effective than increasing timeouts.