Kafka 4 Did Not Move Your Group to KIP-848

Kafka 4.0 did not move your consumer groups onto the new rebalance protocol. It changed one broker default — group.coordinator.rebalance.protocols went from [classic] to [classic, consumer], meaning the broker will now accept a group that speaks the new protocol. The consumer-side switch, group.protocol, still ships with the value classic in every released Kafka through 4.3.1 and on trunk today. Until an application sets it, that application's group is still doing JoinGroup and SyncGroup exactly as it did in 2019.

KIP-848 — Kafka Improvement Proposal 848, "The Next Generation of the Consumer Rebalance Protocol" — is the design that replaces both of those calls with a single ConsumerGroupHeartbeat request. The thing people expect from it is "rebalances get faster." That is not what it does. It removes the idea that a rebalance is one event the whole group participates in.

Below is a group of six consumers reading 24 partitions when a seventh consumer starts up — the deploy that adds a pod. One of the six, consumer-0, is slow: it takes 90 seconds to get back to poll() because it is in the middle of a batch. Move rebalance protocol through its three settings and watch the black bars — a black bar is a partition with nobody fetching from it — and watch which consumer is doing the waiting.

At the settings it loads with, classic plus RangeAssignor reports 1778.2 partition-seconds with no consumer, 0 of 24 partitions never interrupted, and a worst single partition of 89.6 s. Switch to CooperativeStickyAssignor and it reports 1779.1 — very slightly worse. Switch to group.protocol=consumer and it reports 5.6 partition-seconds, with 21 of 24 partitions never interrupted at all. That is a factor of 317, and it is not because anything got faster. Look at which rows are black in the first two settings: all of them except consumer-0's. The slow consumer is the only one still working. Everybody else stopped, to wait for it.

The switch you have not flipped

There are two independent enablements and the release notes talk mostly about the first one.

The broker-side one is group.coordinator.rebalance.protocols. In Kafka 3.9 its default was the single-element list [classic], and the config's own documentation said the consumer protocol "is in early access and therefore must not be used in production." In 4.0 the default became [classic, consumer]; in 4.1 streams was added. In 4.3 the config is deprecated — its documentation now reads "This configuration is deprecated and will be removed in Kafka 5.0. In Kafka 5.0, all protocols will always be enabled and cannot be disabled via this configuration. Use feature versions (group.version, streams.version, share.version) managed by kafka-features.sh instead." So on a modern broker the new protocol is available and you did not have to do anything.

The client-side one is group.protocol, and it has never defaulted to anything but classic. In clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java, on branch 3.7 where the config was introduced, on 4.0, on 4.3, and on trunk today, the line is identical:

public static final String DEFAULT_GROUP_PROTOCOL = GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT);

This is worth being blunt about because a great many write-ups say the opposite. "Kafka 4.0 makes KIP-848 the default" is false. The broker default changed; the consumer default did not. A cluster can be entirely on 4.3 with every consumer group still running JoinGroup and SyncGroup, and nothing in the broker logs will complain, because a classic group on a 4.x broker is a perfectly supported configuration.

The two enablements can also be wrong in the other order, which produces the error people actually search for. Set group.protocol=consumer against a broker that has not enabled it and the client fails immediately with a message that names its own fix:

ConsumerGroupHeartbeatRequest failed due to unsupported version response on
broker side: The cluster does not support the new CONSUMER group protocol.
Set group.protocol=classic on the consumer configs to revert to the CLASSIC
protocol until the cluster is upgraded.

To find out which protocol a group is on right now, ask the broker rather than reading your own YAML — and ask it correctly, because the obvious command does not answer the question. Bare kafka-consumer-groups.sh --bootstrap-server … --list prints group ids one per line with no header and no type. You have to add --type: listGroups() sets includeType = opts.options.has(opts.typeOpt), and only that flag adds the TYPE column. Passing --type with no value filters nothing and prints the column for every group, which is what you want. --describe --state is no help here at all — it prints GROUP, COORDINATOR (ID), ASSIGNMENT-STRATEGY, STATE and #MEMBERS, and no protocol type. Classic means JoinGroup and SyncGroup; Consumer means ConsumerGroupHeartbeat. Nothing else is authoritative — in particular, a group's members can be mixed. The coordinator supports a group where some members speak the classic protocol and some speak the new one, which is what makes a rolling migration possible, and it is also why "I set the config on one service and the pauses did not change" is such a common report.

Why the cooperative assignor did not help either

Set the simulation to CooperativeStickyAssignor and the number gets slightly worse: 1779.1 partition-seconds against 1778.2. Most people find this surprising, because incremental cooperative rebalancing is sold as the thing that stops the group from stopping.

It does do what it says. It does not revoke partitions before the join. In ConsumerCoordinator.onJoinPrepare, a cooperative assignor skips the blanket onPartitionsRevoked that an eager assignor performs, and each member keeps everything the new assignment lets it keep. What it does not change is the barrier, and the barrier is what costs the 89 seconds.

The barrier lives in AbstractCoordinator.joinGroupIfNeeded:

final RequestFuture<ByteBuffer> future = initiateJoinGroup();
client.poll(future, timer);
if (!future.isDone()) {
    // we ran out of time
    return false;
}

That runs on the application thread, inside consumer.poll(). From the moment a member sends JoinGroup until SyncGroup comes back, it is sitting in client.poll(future, timer) and it returns no records to your code. A partition it is keeping is fetched from exactly as little as a partition it is losing. The coordinator does not answer SyncGroup until every member has joined, and consumer-0 will not join until it gets back to poll() 90 seconds from now, so all six consumers spend most of those 90 seconds parked.

So the cooperative assignor removes revocation, not waiting. That is genuinely valuable for anything with per-partition state to rebuild — a Kafka Streams task with a local store, a consumer holding an open file per partition — because it stops that state being thrown away and reconstructed on every deploy. It just does not shorten the pause, and if you adopted it hoping to shorten the pause, the number you measured afterwards was correct and your explanation of it was not.

One target assignment, seven independent walks

Under KIP-848 nothing waits for anything. Here is the whole mechanism.

The group epoch is a counter on the coordinator that increments every time the group's membership or subscription changes. When it increments, the coordinator — not an elected client leader — computes a complete assignment of every partition to every member, using a server-side assignor (UniformAssignor by default; a client may ask for a different one by name with group.remote.assignor). That result is the target assignment, and it is stamped with the epoch it was computed at. It is written to __consumer_offsets like everything else the group coordinator owns, so it survives a coordinator failover.

Each member has a member epoch of its own. A member is caught up when its member epoch equals the target assignment epoch. Reconciliation is the act of moving one member from its epoch to the target epoch, and it happens inside that member's own heartbeat exchange, independently of every other member. The logic is a state machine in group-coordinator/…/modern/consumer/CurrentAssignmentBuilder.java, and it has exactly three outcomes.

The member owes nothing. Its current partitions are a subset of its target, and every partition it is owed is currently unowned. It goes straight to the target epoch and to state STABLE, and the new partitions arrive in the same heartbeat response. It never stopped fetching. In the default run three of the six consumers are in this position and receive no assignment in their heartbeat response at all, because nothing about them changed — under the classic protocol those three would still have had to rejoin and block.

The member owes a partition. Its current assignment is not a subset of the target, so something has to be handed over first. The member does not advance its epoch. It stays where it is, in state UNREVOKED_PARTITIONS, and the heartbeat response tells it the reduced set. The client runs onPartitionsRevoked for just those partitions on the application thread, and then — this is the part that makes the numbers small — sends a heartbeat immediately rather than waiting for the interval, because AbstractMembershipManager.shouldHeartbeatNow() returns true in the ACKNOWLEDGING state. That heartbeat carries a TopicPartitions list that no longer contains the revoked partition, which is how the coordinator learns the handover is done: ownsRevokedPartitions checks precisely that. Only then does the coordinator release the partition and advance the member's epoch.

The member is owed a partition somebody else still holds. It advances to the target epoch anyway and sits in state UNRELEASED_PARTITIONS, fetching everything it already has, and collects the missing partition on a later heartbeat once the current owner has let go. It does not block, and it does not delay anybody.

Put those three together and the group has no synchronisation point. There is no moment when all seven consumers are stopped, because there is no message every consumer has to send before any consumer can proceed. The default run settles in 5.10 s and the 21 partitions that were not moving were never touched.

The wire format makes the same point. ConsumerGroupHeartbeatRequest is API key 68 and carries a MemberEpoch whose documentation reads "0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin". Joining, leaving and heartbeating are one request with a different integer in it. There is no JoinGroup, no SyncGroup, no group leader and no client-side assignor — which is also why partition.assignment.strategy is silently irrelevant once you set group.protocol=consumer, and why a monitoring exporter that reads classic group state reports your healthy group as empty or dead.

Where it stops helping

Three boundaries, all reachable from the controls.

A slow consumer that is gaining still delays its own new partitions. Switch what happened at t=0 to the last consumer shuts down cleanly, keeping KIP-848. The total goes from 5.6 to 98.5 partition-seconds and the worst single partition is 90.8 s — while the classic protocols report 1782.9 s for the same event. Nobody has to revoke anything here, so no member ever pauses; but consumer-0 is one of the four members picking up the orphaned partitions, and onPartitionsAssigned runs on the application thread, the same thread that is 90 seconds deep in a batch. KIP-848 does not give you a way around your own poll loop. A partition assigned to a consumer that is not calling poll() is not being read, whatever the coordinator thinks.

A consumer that owes a partition and cannot produce it gets fenced, and loses everything. Set partitions to 21 — which makes the quotas come out so that consumer-0 is one of the members that must hand a partition over — and then push consumer-0 time back to poll() to 400 s, past the 300 s max.poll.interval.ms. The log prints the real coordinator behaviour: a member in UNREVOKED_PARTITIONS has a timer set to its own rebalance timeout, which the client fills in from its max.poll.interval.ms, and GroupMetadataManager.scheduleConsumerGroupRebalanceTimeout fences it when the timer fires:

log.info("[GroupId {}] Member {} fenced from the group because " +
        "it failed to transition from epoch {} within {}ms.", …);

Fenced means removed. It loses all four of its partitions, not the one it owed, and has to rejoin from epoch 0. Note what the timer is attached to: only a member in UNREVOKED_PARTITIONS is on that clock at all. A member that owes nothing can be as slow as it likes and nobody minds. Under the classic protocol the same 400-second batch removes the member and holds the whole group for the full 300 seconds first — 5076.2 partition-seconds against 27.1.

The new protocol reacts more slowly to a healthy change. Drag consumer-0 time back to poll() down to 0 — every consumer instantly responsive. Classic eager now reports 24.4 partition-seconds with a worst partition of 2.0 s; KIP-848 reports 5.6 with a worst partition of 2.5 s. Lower total, longer worst case. The reason is that group.consumer.heartbeat.interval.ms defaults to 5000 ms where the classic heartbeat.interval.ms defaults to 3000 ms, so a member learns about the change up to two seconds later. Drag group.consumer.heartbeat.interval.ms to 15 s, the maximum the broker will accept, and the "last consumer shuts down" case goes from 98.5 to 115.2 partition-seconds. This interval is set by the broker, not the client: the client sends whatever it likes and the coordinator replies with the interval it must use, clamped between group.consumer.min.heartbeat.interval.ms (5000) and group.consumer.max.heartbeat.interval.ms (15000).

A fourth boundary is not in the simulation because it is not about timing. Any tool that reads consumer group state through the old DescribeGroups path — an exporter scraping lag, a UI, an alerting rule — sees a group that has moved to group.protocol=consumer as something it does not recognise. Real reports include groups rendered as DEAD in a Kafka UI while consuming normally, and lag exporters returning nothing for a Stable group. The new path is ConsumerGroupDescribe. Check that your lag monitoring speaks it before you migrate, not after, because the failure mode is a dashboard that reads zero rather than one that reads an error.

Reading it on a real cluster

Which protocol is this group actually on.

$ kafka-consumer-groups.sh --bootstrap-server broker:9092 --list --type
GROUP            TYPE
orders-indexer   Classic
orders-enricher  Consumer

--type is doing real work there. Without it the same command prints a bare list of group ids and tells you nothing about the protocol, and --describe --state does not print the type either. If the column says Classic, group.protocol is unset or set to classic in that application, whatever the cluster version is.

Whether a migration is stuck half-done. A group can hold classic and consumer members at the same time, and the tool has a column that exists only in that situation. --describe --members --verbose normally prints CURRENT-EPOCH, CURRENT-ASSIGNMENT, TARGET-EPOCH and TARGET-ASSIGNMENT. It adds a fifth, UPGRADED, only when hasClassicMember && hasConsumerMember — so the column appearing at all is the diagnosis. While it is there the group is still doing classic rebalances and you have the cost of the migration and none of the benefit. Finish the rollout and the column disappears.

Whether a member is stuck owing a partition. The coordinator logs the transition at DEBUG with the state name in it: Member … new assignment state: epoch=…, previousEpoch=…, state=UNREVOKED_PARTITIONS. A member that appears there repeatedly, or that you later see in Member … fenced from the group because it failed to transition from epoch N within 300000ms, is not a Kafka problem. It is a consumer whose onPartitionsRevoked callback, or the offset commit inside it, is taking longer than max.poll.interval.ms. Raising the timeout hides it; shrinking max.poll.records fixes it.

On the client. The interesting log line is the state transition from ConsumerMembershipManager, which prints the epoch: Member … with epoch 0 transitioned from UNSUBSCRIBED to JOINING. If you never see an epoch above 0, the heartbeat is being rejected — look for UNSUPPORTED_ASSIGNOR (you asked for a group.remote.assignor the broker does not have) or UNRELEASED_INSTANCE_ID (another process is already using your <code>group.instance.id</code>).

What to measure before and after. Not rebalance count and not rebalance duration — under KIP-848 there is no group-wide rebalance to time, so both metrics become misleading. Measure the thing the simulation measures: per-partition consumption gaps. records-lag per partition during a deploy, or the age of the newest record you have committed, will show you the black bars directly. A group that used to have every partition go quiet for ninety seconds and now has three partitions go quiet for two has improved by a factor you can defend, and no count of rebalances will show it.

Next: the other way to make a deploy stop causing a rebalance, which composes with this one and predates it by four years — give each consumer a stable identity so a restart is not a membership change at all. And if the pauses you are chasing turn out to be the same consumer being kicked out repeatedly, that is a rebalance storm, which is a different problem with a different fix.