Module 01: Consumer Groups and Rebalancing in Depth

Why this matters

You already know consumer groups share partitions and that adding a consumer triggers a rebalance. What that description hides is the cost: with the default protocol, a rebalance is a stop-the-world event across the entire group. Every consumer gives up every partition, waits, and gets a new assignment. During a rolling deploy of ten instances, you don't get one rebalance — you get up to twenty, each pausing all consumption.

Worse is the failure mode teams actually hit: a consumer that takes slightly too long to process a batch gets ejected from the group, which triggers a rebalance, which delays everyone, which makes the next batch bigger, which takes longer to process, which ejects another consumer. That's a rebalance storm, and it looks exactly like "Kafka is down" while every broker is perfectly healthy.

Concepts

What a rebalance actually is

Partition assignment is negotiated, not assigned by the broker. One broker acts as group coordinator; one consumer is elected group leader and computes the assignment.

1. member joins/leaves/times out
2. coordinator marks the group REBALANCING
3. every member sends JoinGroup      ─┐  all consumption stopped
4. leader computes assignment         │  (eager protocol)
5. every member sends SyncGroup      ─┘
6. members receive partitions, resume

Steps 3-5 are the stall. With the eager protocol, all members revoke all partitions before step 3, so the group processes nothing until step 6 completes — for every member, even those whose assignment doesn't change.

The four timeouts, and which one bites

The distinction that matters: heartbeats come from a background thread, so a consumer stuck processing records still looks alive to the coordinator. max.poll.interval.ms is what catches it — it measures the gap between poll() calls, which only advances when you finish processing the previous batch.

So the equation you must satisfy is:

max.poll.records × per-record processing time  <  max.poll.interval.ms

500 records × 700 ms each = 350 s, against a 300 s default. That consumer gets ejected mid-batch, its partitions get reassigned, the records are reprocessed by someone else, and — because the ejected consumer is still working — you now have two consumers processing the same partition. This is the single most common Kafka consumer bug in production.

The fix is almost always lower max.poll.records, not raise the timeout. Smaller batches poll more often, which both keeps you in the group and makes each rebalance cheaper.

Eager vs cooperative-sticky

EAGER (default, RangeAssignor/RoundRobinAssignor)
  before:  C1[p0,p1]  C2[p2,p3]
  C3 joins ──▶ ALL revoke everything ──▶ full stop ──▶ reassign
  after:   C1[p0]  C2[p2]  C3[p1,p3]      every consumer was paused

COOPERATIVE STICKY
  before:  C1[p0,p1]  C2[p2,p3]
  C3 joins ──▶ only p1 and p3 revoked ──▶ C1 keeps p0, C2 keeps p2 throughout
  after:   C1[p0]  C2[p2]  C3[p1,p3]      only moved partitions paused

{"partition.assignment.strategy": "cooperative-sticky"}

Cooperative rebalancing (KIP-429) revokes only the partitions that must move, in two rounds, so consumers keep processing everything they retain. On a large group this is the difference between a multi-second full stop and a barely-visible blip.

Two caveats worth knowing before you switch it on:

• You cannot mix strategies within a group. Migrating a running group requires a two-step rolling deploy (first deploy supporting both, then switch), or a full stop. Changing it in one rolling deploy will fail the group.
• Your on_revoke handler must not assume it's losing everything — with cooperative it receives only the revoked subset.

Static membership: stop rebalancing on restarts

By default, a restarting consumer is a new member with a new ID, so a rolling restart triggers a rebalance per instance — twice, once for the leave and once for the join.

{"group.instance.id": "consumer-3"}      # stable, unique per instance

With a group.instance.id (KIP-345), the member keeps its identity and its partition assignment across a restart. If it comes back within session.timeout.ms, no rebalance happens at all. In Kubernetes this maps naturally onto a StatefulSet's ordinal pod name.

The tradeoff: a genuinely dead static member isn't detected until session.timeout.ms expires, so its partitions stall for that long. Static membership trades faster failure detection for far fewer rebalances — usually a good trade for large groups doing frequent deploys.

Rebalance listeners: where you commit

from confluent_kafka import Consumer

def on_assign(consumer, partitions):
    log.info("assigned: %s", [(p.topic, p.partition) for p in partitions])

def on_revoke(consumer, partitions):
    # last chance to commit — after this we no longer own these partitions
    try:
        consumer.commit(asynchronous=False)
    except Exception as e:
        log.warning("commit on revoke failed: %s", e)

consumer.subscribe(["orders"], on_assign=on_assign, on_revoke=on_revoke)

on_revoke fires before partitions are taken away — it is the only safe place to flush in-progress work and commit offsets. Skipping it means the next owner restarts from the last committed offset and reprocesses everything since.

subscribe() vs assign()

consumer.subscribe(["orders"])                        # group-managed, rebalances
consumer.assign([TopicPartition("orders", 0)])        # manual, no group, no rebalance

assign() opts out of group management entirely: no rebalances, no automatic failover, and you own partition distribution. It's right for single-instance tools, replaying a specific partition, or systems with their own assignment logic — and wrong for anything that needs to scale elastically.

Partition count is your parallelism ceiling

6 partitions, 3 consumers  ->  2 each                      ✓
6 partitions, 6 consumers  ->  1 each                      ✓
6 partitions, 8 consumers  ->  6 working, 2 idle forever   ✗

Extra consumers beyond the partition count do nothing — they sit idle holding a group slot. Since module 00 established that adding partitions to a keyed topic breaks per-key ordering, partition count is effectively a capacity decision you make up front. Over-provision modestly (a topic with 12 partitions and 3 consumers scales to 12 later); don't over-provision wildly, since every partition costs broker file handles, memory and replication traffic.

Command reference

Hands-on exercises

Use the KRaft broker from module 00, and:

kt --create --topic rebal --partitions 6
kg() { docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 "$@"; }

A reusable consumer script, consumer.py:

import sys, time, logging
from confluent_kafka import Consumer

logging.basicConfig(level=logging.INFO, format=f"[{sys.argv[1]}] %(message)s")
log = logging.getLogger()

conf = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "demo",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": True,
}
if len(sys.argv) > 2 and sys.argv[2] == "coop":
    conf["partition.assignment.strategy"] = "cooperative-sticky"
if len(sys.argv) > 3:
    conf["group.instance.id"] = sys.argv[3]

c = Consumer(conf)
c.subscribe(["rebal"],
            on_assign=lambda _, ps: log.info("ASSIGNED %s", sorted(p.partition for p in ps)),
            on_revoke=lambda _, ps: log.info("REVOKED  %s", sorted(p.partition for p in ps)))
try:
    while True:
        msg = c.poll(1.0)
        if msg and not msg.error():
            log.info("p%d @%d", msg.partition(), msg.offset())
except KeyboardInterrupt:
    pass
finally:
    c.close()

1. Watch an eager rebalance revoke everything

Start one consumer, wait for assignment, then start a second, then a third:

python consumer.py C1 &
sleep 5 && python consumer.py C2 &
sleep 5 && python consumer.py C3 &

Expected — C1 gives up all six partitions, then gets three back:

C1  ASSIGNED  [0, 1, 2, 3, 4, 5]
--  C2 JOINS
C1  REVOKED   [0, 1, 2, 3, 4, 5]     <-- all six, including ones it keeps
C1  ASSIGNED  [1, 3, 5]
C2  ASSIGNED  [0, 2, 4]

C1 surrendered partitions 1, 3 and 5 only to be handed them straight back. The gap between REVOKED and ASSIGNED is dead time for the whole group — and it scales with group size, because every member does this.

2. Compare with cooperative-sticky

Kill everything, then repeat with the coop argument.

Expected — C1 revokes only the half that actually moves:

C1  ASSIGNED  [0, 1, 2, 3, 4, 5]
--  C2 JOINS
C1  REVOKED   [0, 1, 2]      <-- only what moves; C1 keeps 3,4,5 the whole time
C1  ASSIGNED  []             <-- round 2 of the protocol: nothing new for C1
C2  ASSIGNED  []
C2  ASSIGNED  [0, 1, 2]

Two things to notice. C1 never stopped serving partitions 3, 4 and 5 — compare that to exercise 1, where it dropped all six. And the empty ASSIGNED [] callbacks are normal: cooperative rebalancing runs in two rounds, so you'll see callbacks that convey no change. Handlers must tolerate an empty partition list rather than assuming every callback is meaningful.

3. Prove you can't mix strategies

With a cooperative consumer running, start an eager one in the same group.

Expected: the group fails to stabilise, with an inconsistent-protocol error. Write down the error — it's what a careless rolling deploy produces.

4. Show extra consumers sit idle

Start 8 consumers against the 6-partition topic.

Expected: exactly 2 log an empty assignment. Confirm with:

kg --describe --group demo --members

5. Trigger ejection with slow processing

Run one consumer with:

{"max.poll.interval.ms": 10000, "max.poll.records": 500}

and a handler that sleeps 50 ms per record. Produce 1000 records.

Expected: after ~10 s the consumer is kicked out mid-batch (MaxPollExceeded / the group rebalances), then rejoins, and reprocesses records it had already handled. Log a set of processed IDs and confirm duplicates. Then fix it by setting max.poll.records=20 and confirm the ejection stops — without touching max.poll.interval.ms.

6. Static membership eliminates restart rebalances

Run three consumers with group.instance.id values c1, c2, c3. Wait for a stable assignment, then kill c2 and restart it within session.timeout.ms.

Expected: no REVOKED/ASSIGNED lines on c1 and c3 — they never noticed. Repeat without group.instance.id and observe two full rebalances.

7. Commit on revoke, or reprocess

Disable auto-commit, commit only in on_revoke, and process records with a visible counter. Trigger a rebalance by starting a second consumer.

Expected: no reprocessing. Now remove the commit from on_revoke and repeat.

Expected: the new owner replays from the last committed offset, reprocessing records the first consumer had already handled.

8. Diagnose and fix: the consumer group that never caught up

A team's payments consumer group has 12 instances against a 12-partition topic. Lag grows steadily during business hours and never recovers. Logs show constant ASSIGNED/REVOKED cycling. Their config:

{"group.id": "payments", "max.poll.records": 500,
 "max.poll.interval.ms": 300000, "session.timeout.ms": 45000,
 "enable.auto.commit": True, "auto.commit.interval.ms": 5000}

Each record makes a ~400 ms external payment-API call. Scaling to 24 instances made it worse.

9. Clean up

kg --delete --group demo

Independent challenge

No solution given. You run a consumer group processing image-thumbnail jobs: 30 partitions, 30 instances on Kubernetes, each record takes 2-45 seconds (highly variable — small photos vs. large RAW files). Deploys happen 5-10 times a day and currently cause a 4-6 minute processing stall. Occasionally a single enormous file takes 8 minutes and ejects its consumer.

Design a configuration and deployment strategy that keeps deploys under 30 seconds of disruption and stops the ejections, without losing at-least- once delivery. Then address the harder question: max.poll.records=1 still doesn't help the 8-minute record, so what do you actually do about work items whose duration exceeds any reasonable poll interval?

Common mistakes & troubleshooting

• Raising max.poll.interval.ms instead of lowering max.poll.records. It masks the symptom and delays detection of genuinely dead consumers.
• Assuming heartbeats prove liveness. They come from a background thread; a consumer stuck in processing still heartbeats. max.poll.interval.ms is the setting that catches it.
• Mixing assignment strategies in one group. The group won't stabilise — migrating to cooperative needs a two-phase rollout.
• More consumers than partitions. The extras idle forever and add rebalance cost.
• No on_revoke commit. Every rebalance reprocesses everything since the last commit.
• Assuming on_revoke means "all partitions" — under cooperative it's only the moved subset.
• Auto-commit with slow or fallible processing. Offsets advance on a timer regardless of whether the work succeeded.
• Not using group.instance.id on a frequently-deployed service, paying two rebalances per instance per deploy.
• Trying to scale throughput by adding consumers when the partition count is the actual ceiling.

Checkpoint quiz

Further reading & sources

• KIP-429: Incremental cooperative rebalancing - the design and migration path for cooperative-sticky.
• KIP-345: Static membership - why group.instance.id removes restart rebalances.
• Kafka: Consumer configuration reference - defaults and semantics for every timeout discussed here.
• Confluent: Consumer group protocol - the JoinGroup/SyncGroup sequence walked through visually.
• confluent-kafka-python: Consumer - subscribe, assign, pause, and the rebalance callbacks.

Next

02-delivery-semantics-and-exactly-once — you can now keep a group stable. Module 02 addresses what those consumers actually guarantee about processing each record once.