The Group Coordinator Does Not Assign Your Partitions

There is no cluster-wide service that decides which consumer reads which partition. There is one broker, picked by hashing the characters of your group's name, and under the protocol Kafka still ships as the default that broker does not compute an assignment at all.

The group coordinator is the broker that holds membership and committed offsets for one consumer group. Two questions about it have answers most people have never had to give: which broker is it, and what does it decide. The first answer is a hash of the group id. The second answer, for the classic protocol, is: almost nothing.

Start with the first. A consumer that wants to join a group sends a FindCoordinator request naming the group. Any broker can answer it, and every broker computes the same three lines. The panel below runs those lines. Type a group name you actually use.

At the defaults, orders-service hashes to 23943117, which is partition 17 of 50, which broker 2 leads. That is the entire coordinator election. Now change the group name to orders-service-v2: the hash becomes 324874364 and the partition becomes 14. Change it again to orders-service-v3 and the hash goes up by exactly one, the partition goes to 15, and the coordinator moves to broker 0. One character in a name your deployment tool generated moved your group to a different machine.

Look at the bar chart while you do it. Twenty-four groups across three brokers should be eight each. It is 10, 6 and 8. Drag brokers in cluster to 9 and it gets worse, not better: brokers 2 and 6 coordinate five groups each and broker 7 coordinates none at all. You added six machines and the group traffic did not spread, because nothing in this mechanism is trying to balance anything. It is a modulo.

Where the coordinator is, exactly

Three lines of Kafka decide it. In GroupCoordinatorService:

@Override
public int partitionFor(String groupId) {
    throwIfNotActive();
    return Utils.abs(groupId.hashCode()) % numPartitions;
}

numPartitions is offsets.topic.num.partitions, which defaults to 50. Its own config documentation says it should not change after deployment, and the reason is this line: change the modulus and every group in the cluster rehashes to a different partition, so every group's coordinator moves and every committed offset is looked for in the wrong place.

Utils.abs is not Math.abs: its javadoc says if the number is Int.MinValue return 0. This is different from java.lang.Math.abs, because Math.abs(Integer.MIN_VALUE) returns Integer.MIN_VALUE and would produce a negative partition. Exactly one 32-bit hash in four billion needs that branch, and Kafka has it.

The coordinator is then the leader replica of that partition. This is the sentence with all the consequences in it, because partition leadership is not a stable property of a broker. It moves on a rolling restart, on a preferred-leader election, on a disk failure, and on any reassignment. When it moves, the group's coordinator moves with it, and every client holding a connection to the old one gets NOT_COORDINATOR and has to send FindCoordinator again.

That is the mechanism behind the log line people file bugs about. Marking the coordinator dead is not a broker crash. It is a client discovering that the broker it was talking to is no longer the leader of one partition of an internal topic. The librdkafka issue titled Kafka consumer gets stuck (NOT_COORDINATOR) whilst rejoining a group after a broker rolling update ran for nineteen comments and eight months, and the shape of it is always the same: a rolling restart moves 50 partition leaderships, which moves every group's coordinator, and every consumer in the cluster re-discovers at once.

Take broker 0 offline in the panel with offsets.topic.replication.factor at 3 and read the log. Ten of the twenty-four groups had their coordinator there; all ten move to another replica and nothing breaks. Now drag the replication factor to 1 and take broker 0 offline again. The same ten groups hash to a partition whose only replica is gone, and they get COORDINATOR_NOT_AVAILABLE. They cannot join, they cannot heartbeat, and they cannot commit an offset — with healthy data topics and healthy consumers. This is why the single-broker development default of offsets.topic.replication.factor=1 is the most dangerous setting to carry into a real cluster: it is invisible until the first restart.

One more state exists between those two. When a broker becomes leader of an offsets partition it has to read the whole partition back into memory before it can answer for those groups, and while it does it returns COORDINATOR_LOAD_IN_PROGRESS. On a cluster with large __consumer_offsets segments this is the pause after a restart where the brokers are up, the topics are readable, and no consumer group can commit anything.

The broker does not run your assignor

Now the second question. Here is the class javadoc of AbstractCoordinator, the client-side base class for group membership, describing the protocol in its own words:

Kafka's group management protocol consists of the following sequence
of actions:

1. Group Registration: Group members register with the coordinator
   providing their own metadata (such as the set of topics they are
   interested in).
2. Group/Leader Selection: The coordinator select the members of the
   group and chooses one member as the leader.
3. State Assignment: The leader collects the metadata from all the
   members of the group and assigns state.
4. Group Stabilization: Each member receives the state assigned by the
   leader and begins processing.

Read who does what. The coordinator picks the members and picks a leader. A member — one of your consumer processes — collects the metadata and assigns the partitions. The broker runs no assignor. In ClassicGroup.add() the election is one branch: if (leaderId.isEmpty()) leaderId = Optional.of(member.memberId()). The first member to join is the leader. Not the oldest, not the least loaded, not one you can choose — the first to send JoinGroup after the group emptied.

The round trip is then: every member sends JoinGroup with its subscription; the coordinator replies to the leader with all of the members' subscriptions and to everyone else with an empty list; the leader calls onLeaderElected, whose javadoc is invoked when the leader is elected. This is used by the leader to perform the assignment if necessary and to push state to all the members of the group; and it sends the finished assignment back in SyncGroup. The coordinator stores those bytes and hands each member its own share. It does not parse them.

Three practical consequences follow from that, and none of them are obvious from the outside:

• A lopsided assignment is a bug in one of your pods, not in the cluster. The RangeAssignor skew described in consumer groups and partition assignment is computed by a JVM you deployed.
• If the elected leader is the slow pod, the whole group waits for it to return from your code before anyone gets an assignment. Which pod that is is decided by join order.
• The assignor is negotiated between clients, so you cannot change it in one deploy. That one has a mechanism worth watching.

Why changing the assignor takes two deploys

The coordinator has to pick one assignor name for the group, from lists the members supply. ClassicGroup.selectProtocol() states the rule: Each member will vote for a protocol and the one with the most votes will be selected. Only a protocol that is supported by all members can be selected. A candidate protocol is one whose supporting-member count equals the group size; each member then votes for the first entry of its own partition.assignment.strategy list that is still a candidate.

There is a second rule underneath it, and it is the one that bites. Before a member is allowed into the group at all, ClassicGroup.supportsProtocols() asks whether any protocol the member offers is already supported by every member currently in the group. If none is, the coordinator answers that member's JoinGroup with INCONSISTENT_GROUP_PROTOCOL and leaves the existing group untouched. The pod that cannot join stops consuming. The pods that never left keep consuming everything.

That is an admission test followed by an intersection followed by a vote, and it behaves nothing like a setting. Run a rolling deploy through it.

Leave it on step 2 and drag pods rolled from 0 to 6. The group runs range at zero pods. At one pod — the first restart, five of six pods still on the old config — the protocol is already cooperative-sticky. It does not wait for a majority, because it is not a majority vote over preferences; it is a vote over the intersection, and the moment one member stops offering range the intersection has one element in it. All six pods stay in the group throughout, because all six still offer cooperative-sticky. That is the whole point of having done the first bounce: it put cooperative-sticky on every member's list so that the second bounce can flip the group on its first restart without locking anybody out.

Now switch to one step, which is what everybody tries first: take a group running plain [range] and set the new pods to [cooperative-sticky]. Watch pods in the group rather than the protocol. It goes 6, 5, 4, 3, 2, 1, and then back to 6. Every pod you restart is refused at JoinGroup and sits in a retry loop, while the pods you have not touched yet keep all the partitions. At five of six rolled, one pod is consuming the entire topic and the other five are idle. Consumption does not stop, which is exactly why this is hard to spot: lag climbs, the group reports Stable, and --describe shows a shrinking membership that looks like pods failing to start. The group only accepts the new configuration when the last old pod leaves and the group goes Empty — at that moment the admission test has no members to intersect against and lets the first newcomer set the protocol.

Move members in group anywhere from 2 to 12 and the floor is always one pod. The size of the group does not soften it; a twelve-pod deploy spends its eleventh step running the whole topic through a single consumer.

The CooperativeStickyAssignor javadoc gives the rule without the mechanism: To turn on cooperative rebalancing you must set all your consumers to use this PartitionAssignor, and if upgrading from 2.3 or earlier, you must follow a specific upgrade path. The mechanism is the admission test. And it is why Kafka 3.0 changed the client default to [range, cooperative-sticky] rather than to [cooperative-sticky]: shipping the two-element list to everybody put every group one bounce away from cooperative instead of two, and made the dangerous one-step path something you have to opt into.

Select step 1 and drag the slider across. Nothing happens at all: six pods in the group at every step, range throughout, and the protocol never flips. That is the correct outcome and the reason the first bounce is easy to skip — it is a deploy with no observable effect whose only job is to make the next one safe.

The add a custom assignor plan shows the same rule from the other side. New pods list a custom assignor first and keep range behind it, so they are admitted at every step. The group keeps running range until the last old pod is gone, then flips to the custom assignor at pod 6. A preference at the front of the list is worth nothing until the last member also has it. team-sticky is invented for this panel and is shown as cooperative because that is what a custom assignor returning RebalanceProtocol.COOPERATIVE from supportedProtocols() would do; the name is not a real Kafka assignor.

Checking it in a running system

Four checks, in the order that narrows fastest.

Which broker. kafka-consumer-groups.sh --bootstrap-server … --describe --group g --state prints a COORDINATOR (ID) column. Compare it against the hash: if you know your group id you can compute the partition yourself and check that the leadership of __consumer_offsets-<n> agrees. When several groups misbehave at once, hash all of them; if they share a partition or a broker, you have a coordinator problem rather than a consumer problem.

Whether the offsets topic can survive a restart. kafka-topics.sh --describe --topic __consumer_offsets. Any partition whose Isr list is shorter than its Replicas list is a group outage waiting for the next restart. At replication factor 1 each broker is the only replica for its share of the 50 partitions, so losing one broker out of n puts roughly one group in n into COORDINATOR_NOT_AVAILABLE — on a three-broker cluster, about a third of them at once. You cannot raise this with an alter on the config; the topic is already created, so it takes a partition reassignment.

Which client is the leader. Add --members --verbose to the --describe call and you get every member id, host and assignment. The group leader is not marked, but it is the member whose logs contain Finished assignment for group at generation after the last rebalance. ConsumerCoordinator.onLeaderElected logs that line at INFO, so it is there without changing any log level, and it appears in exactly one pod. Grep your fleet for it. Do not search for Performing assignment using strategy, which sits three lines above it in the same method: that one is log.debug and you will not see it at a default configuration. If the pod the INFO line names is also the one with the highest processing latency, the group's assignment latency is that pod's latency.

What protocol the group actually settled on. The --state output has an ASSIGNMENT-STRATEGY column, and it is the coordinator's answer, not your configuration file's. If it says range after you rolled out cooperative-sticky, one member is still offering an older list and the intersection has been decided by that member. If the group is stuck in PreparingRebalance and the client logs show INCONSISTENT_GROUP_PROTOCOL, the intersection is empty and you are mid-way through a one-step assignor change.

Whether a deploy is quietly shrinking the group. This is the one the panel above was built for, and it does not look like an error. Every time a classic group settles, the coordinator broker logs Stabilized group <id> generation <n> with <m> members at INFO. Watch <m> across a rolling deploy. If it counts down — 6, 5, 4 — while your pods are all reporting healthy, the restarted pods are being refused at JoinGroup and the group is running on the ones you have not touched yet. Grep those pods for INCONSISTENT_GROUP_PROTOCOL; the client logs the rejection and retries silently forever, so nothing crashes and nothing restarts. A member count that recovers to full only after the last pod restarts confirms it.

What none of this fixes is the division of labour itself: an elected client computing the assignment, a global barrier every member has to reach, and a broker that can only wait. That is what the KIP-848 consumer group protocol rearranges — KIP-848 is Kafka Improvement Proposal 848 — by moving the assignor onto the coordinator and reconciling members one at a time. It reached general availability in Kafka 4.0, where GroupVersion.GV_1 (Version 1 enables the consumer rebalance protocol) became the latest production feature level and group.coordinator.rebalance.protocols gained consumer in its default. So the broker is ready and the consumer is not: group.protocol still defaults to classic on the 4.3 client, and you have to set it to consumer yourself. When you do, partition.assignment.strategy stops being a valid consumer config at all — ConsumerConfig lists it under CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS — because the vote this lesson just walked through no longer exists. Before that, the two clocks the coordinator runs are worth understanding on their own — what a rebalance actually costs — and if your groups churn on every deploy, static membership changes what the coordinator does when a member disappears.