Deep Dive Into Kafka Rebalancing

In my last Kafka post, I wrote that if you block the listener thread for too long, "Kafka thinks you're dead, triggers a rebalance". That's not quite right. Kafka knows you're alive the whole time. Your consumer is the one that gives up and leaves.

I only found that out because I went digging into what actually happens during a rebalance. Most of us know the high level version. A topic is split into partitions, consumers in a group share those partitions, and when a consumer joins or leaves, Kafka rebalances. But the details are where all the weird production behaviour comes from.

So I read the KIPs and went through the client and broker source. This post is what I learned.

The setup

One example for the whole post. A topic with six partitions, p0 to p5. A consumer group called order-service with three consumers, A, B and C. Each consumer owns two partitions.

The one rule that matters: a partition is only ever owned by one consumer in the group at a time. That's how Kafka keeps the messages in a partition in order. P0 is only read by A, and A reads it in order.

Now say C goes away. Maybe it was shut down, maybe it crashed. P4 and P5 have no owner, and someone has to decide who gets them. That someone is the group coordinator.

The coordinator is just a broker

There's no special coordinator service. It's one of your brokers.

Kafka has an internal topic called __consumer_offsets where groups store their committed offsets. It has 50 partitions by default. Kafka hashes the group name, takes it mod 50, and whichever broker leads that partition becomes the group's coordinator. For order-service, that's partition 26.

When a consumer starts, all it knows is bootstrap.servers. It asks any one of those brokers "who's the coordinator for order-service?" with a FindCoordinator request, gets an address back, and from then on everything for the group goes to that broker: joining, heartbeats, offset commits. If that broker dies or the partition moves, requests start failing with a "not coordinator" error and the consumer just asks again.

A rebalance is two steps

Let's start with the simplest case. order-service is starting up for the first time and nobody is consuming yet.

Step one is JoinGroup. Each consumer sends a JoinGroup request to the coordinator. It's basically "I'd like to join", along with the topics it's subscribed to and the assignment strategies it supports (range, cooperative-sticky and so on).

The coordinator has a small problem here. It's a brand new group, so it has no idea how many consumers are coming. If it started as soon as A joined, A would get all six partitions, then B would join and force another rebalance, then C.

Once it's done waiting, the coordinator replies to everyone and picks a leader. The leader is simply the first consumer that joined, so A. The leader's reply has the full member list and what everyone is subscribed to. B and C just get "you're in, A is your leader".

Now the part that surprised me. The broker doesn't decide who gets which partition. A does. The leader is a regular consumer, and it runs the assignment strategy itself.

Step two is SyncGroup. Every consumer sends a SyncGroup request. A's has the full plan. B's and C's are empty, which is them asking "so what did I get?". The coordinator holds on to B and C until A's plan arrives, then sends each consumer only its own partitions.

That's a rebalance. Everything else in this post is about what triggers one and how long the group stops for.

Heartbeats

Once a consumer is in the group, the coordinator needs to know it's still alive. Every consumer sends a small heartbeat every 3 seconds (heartbeat.interval.ms). If the coordinator doesn't hear from a consumer for session.timeout.ms, 45 seconds by default, it decides the consumer is dead and kicks it out.

Two details that matter later:

1. Heartbeats come from a background thread, not the thread processing your messages. If your consumer is stuck on a slow batch, heartbeats keep going just fine.
2. The broker never sends anything on its own. It only replies to requests. So when a rebalance starts, the coordinator can't go and tell A and B. It waits for their next heartbeat and replies with "rebalance in progress, please rejoin". That's how consumers usually find out.

Three ways to leave a group

Graceful shutdown

C is shutting down properly, say we're scaling down. It calls close(), which runs your revoke callback (where you commit offsets, so whoever gets C's partitions knows where C stopped), and then sends a LeaveGroup request. "I'm leaving, don't wait for me."

The coordinator starts a rebalance right away. A and B find out on their next heartbeat. And this is where the original protocol, called eager rebalancing, does something drastic. Before rejoining, A and B give up all of their partitions. Not just the ones that need to move. All of them.

A had nothing to do with C leaving, but it still stops reading p0 and p1. For a moment, nobody in the group is consuming anything. Then A and B rejoin, A works out a new plan, and SyncGroup hands out p0 to p2 to A and p3 to p5 to B.

The coordinator also waits for every known member to rejoin before it moves on. It won't wait forever (the limit is the rebalance timeout, which for consumers is max.poll.interval.ms, 5 minutes by default), but if one member is slow to rejoin, everyone waits for it.

Crash

C gets OOM-killed. No close(), no commit, no LeaveGroup. The coordinator has no idea C is gone.

The only signal is that C's heartbeats stop. But a heartbeat can be late for lots of reasons, like a GC pause or a network blip, so the coordinator waits the full 45 seconds before it does anything. For that whole time, p4 and p5 just sit there.

And since C never got to commit, whoever picks up p4 and p5 starts from C's last committed offset. Anything C processed after that commit gets processed again.

Too slow between polls

This is the one I got wrong in my last post.

C is alive, but stuck on a slow batch. Maybe a database call is hanging, maybe a vendor API is taking three minutes. Its heartbeats are coming from the background thread, so from the coordinator's side everything looks fine.

What catches it is max.poll.interval.ms, the maximum time allowed between two poll() calls. Five minutes by default. And it's checked by the consumer itself, not the coordinator. Once C goes past it, C realises it's fallen behind and sends a LeaveGroup on its own. Rebalance. P4 and P5 go to someone else.

But C is still working on that batch. When it finishes and tries to commit, the commit fails with a CommitFailedException. It doesn't even reach the broker. The client knows it's not in the group anymore. Meanwhile, the new owner of p4 and p5 processes those same messages again. Then C calls poll(), sends JoinGroup, and that's a second rebalance.

So the two timeouts sound similar but ask different questions:

• session.timeout.ms is the coordinator asking "is this consumer alive?", using heartbeats.
• max.poll.interval.ms is the consumer asking itself "am I getting any work done?", using poll() calls.

A consumer can be perfectly alive and still not be making progress. That's why you need both.

Why deploys hurt

Now put it together with something we do every day. Most deploys are rolling restarts: restart the consumers one at a time so the service stays up.

Each restart is a LeaveGroup (rebalance) and then a JoinGroup when the pod comes back (another rebalance). Three consumers, two rebalances each, six in total. On eager, each of those stops the entire group.

It can be fewer. If the pod comes back fast enough, its JoinGroup arrives while the first rebalance is still running and gets folded into it. But real pods take a while to start, so six is closer to what you'll see.

Static membership

The consumer that's restarting isn't really leaving. It's the same consumer, it'll be back soon, and ideally it should get its old partitions back. That's what static membership does (KIP-345).

You give each consumer a fixed group.instance.id. You set it yourself, because it has to survive restarts. The pod name works well. Every consumer needs a unique one.

A static consumer doesn't send LeaveGroup when it shuts down. If it comes back with the same id before session.timeout.ms runs out, the coordinator recognises it and hands back the partitions it had. No rebalance at all. So you'd usually raise the session timeout to something longer than a restart.

If you don't set group.instance.id, there's no default. It's null, and the consumer is a normal dynamic member that rebalances on every restart.

It isn't free, though:

• While a static consumer is restarting, nobody else picks up its partitions.
• If a static consumer actually crashes, the coordinator only notices after the session timeout, and you just made that longer.
• It only helps when the same consumer comes back. Adding or removing consumers still rebalances.

You're trading how fast you notice a dead consumer for fewer rebalances. For most services that deploy several times a day, that's a good trade.

Cooperative rebalancing

Static membership fixes restarts. But when a real rebalance does happen, eager still stops the whole group. And most of the time there's no need. When a new consumer joins, only a couple of partitions actually have to move.

That's the idea behind cooperative rebalancing (KIP-429). Nobody gives up everything. You keep what you have, and only the partitions that need to move are given up.

Say A has p0 to p2, B has p3 to p5, and C starts up. A and B are told to rejoin, but this time they keep consuming, and their JoinGroup also says which partitions they currently own. A, still the leader, runs the cooperative-sticky assignor: three consumers, six partitions, two each. A keeps p0 and p1, B keeps p3 and p4, and p2 and p5 should go to C.

But A doesn't give them to C yet. The first round only takes them away. A gives up p2, B gives up p5, and everything else keeps flowing. Then A and B immediately rejoin, a second rebalance happens, and this time A sees p2 and p5 have no owner and gives them to C.

Why two rounds? Because a partition should never have two owners at once. If C started on p2 while A was still finishing with it, you'd get duplicates or messages out of order. Round one makes sure the old owner has let go. Round two hands it over.

There's still a catch. The second round is a group-wide rebalance, so it can't finish until every consumer has rejoined. If B is slow to let go of p5 (say it's flushing something in its revoke callback), C has to wait for B before it gets p2 too, even though p2 came from A and has nothing to do with B.

The new protocol in Kafka 4

That catch is what KIP-848 fixes. It became generally available in Kafka 4.0 and it changes how the whole thing works. Three big differences:

1. No JoinGroup or SyncGroup. Consumers just keep sending one request, ConsumerGroupHeartbeat. Joining, staying in, and getting your partitions all happen through it.
2. The broker does the assignment. No leader consumer anymore.
3. No group-wide rounds. Each consumer moves to its new partitions on its own.

Same example. C sends its first heartbeat. The coordinator sees a new member, bumps the group epoch (basically the group's version number), and works out the target: A keeps p0 and p1, B keeps p3 and p4, C gets p2 and p5.

Then it moves everyone towards that target one heartbeat at a time. A's next heartbeat reply says "give up p2". B's says "give up p5". A lets go right away and sends a heartbeat straight after saying it's done. On C's next heartbeat, C gets p2. Nobody waited for B. When B finally finishes, it reports back, and C gets p5 on its next heartbeat.

One trade-off: everything moves on heartbeats, and the default heartbeat interval with the new protocol is 5 seconds, so a single handover can take a few seconds even when nobody is slow. The win isn't that each handover is faster. It's that one slow consumer can no longer hold up the whole group.

And remember when I said the broker doesn't assign partitions, the leader consumer does? That's true for the classic protocol. With the new one, the broker does it. So if someone tells you "the broker assigns partitions", they're right or wrong depending on which protocol you're on.

To switch, set group.protocol=consumer. The session timeout and heartbeat interval become broker settings, partition.assignment.strategy is no longer used, and static membership still works.

What this looks like in production

Here's where the mental model pays off. Three things you'll probably run into.

Lag spikes on every deploy. That's the rolling restart: up to two rebalances per pod, and on eager each one stops the group. To confirm, check the coordinator's broker log during a deploy. On the classic protocol, every rebalance logs a line starting with "Preparing to rebalance group" along with the reason. Count them and match them to your pods. Fix: static membership with a session timeout longer than a restart, plus cooperative-sticky or the new protocol for the rebalances you can't avoid.

The group never settles. Constant rebalancing, CommitFailedException in the logs, the same messages processed again and again. That's the slow-poll loop. The consumer logs a warning that the poll timeout has expired, and even tells you the gap was longer than max.poll.interval.ms. Fix: do less per poll. Lower max.poll.records, make processing faster, or raise max.poll.interval.ms. Just remember that's also your rebalance timeout, so raising it means a stuck consumer can hold up a rebalance for longer. Or do what I did in the inventory consumer and hand the work off to a thread pool, so the poll thread stays free.

A couple of partitions stall for about 45 seconds after a pod dies. That's the crash case. The gap is your session timeout. There's no perfect fix here, only a trade-off. Lower the timeout and crashes get noticed sooner, but a long GC pause can now get a healthy consumer kicked out. And if you're on static membership, you raised this timeout on purpose.

That's rebalancing

A rebalance on the classic protocol is just JoinGroup and SyncGroup. Everything painful about it comes from how often it's triggered and how long it makes the group wait. Static membership cut down how often. Cooperative rebalancing stopped the whole group from pausing. KIP-848 stopped one slow consumer from holding everyone else up.

If your Kafka cluster is on 4.x, try group.protocol=consumer on one service and watch what happens on your next deploy. And if you're still on eager with dynamic members, you're probably eating six group-wide pauses every time you ship.

Next up, I want to dig into partition assignment strategies, and how a group of consumers that each support different ones actually agrees on one.