Digging into Kafka's Source 11: The Consumer and the Group Coordinator — Inside poll() and Rebalancing

What this post covers

This is the eleventh post in the "Digging into Kafka's Source" series. In post 8 we saw how the broker hands out records for a Fetch request, and in post 10 we saw how the controller keeps the cluster's ledger (the metadata log). Now we go to the receiving side: the consumer. Pulling bytes in with Fetch was already covered in post 8, so this post focuses on the problem that only consumers have — "how do several consumers share partitions, and when one joins or leaves, who splits them up again?"

This post zooms in on the consumer inside region A (clients) and the group coordinator inside region F (coordinators) of the post 2 map. Grey boxes belong to earlier posts.

It tries to answer four questions:

• When you create a KafkaConsumer, which class actually does the work, and what does the group.protocol setting change?
• What does the thread that calls poll() do inside it, and who sends the heartbeats?
• When another consumer joins, who starts the rebalance and who computes the partition assignment — and how do the old way (classic) and the new way (consumer) differ?
• Where does the broker's group coordinator keep group state, and how does it get that state back after the broker restarts?

All line numbers in this post refer to this commit. How to get the same source and build the example Docker image is in post 1. The classes read here are nearly identical to the latest stable release most people use, 4.4.0. Comparing with the 4.4.0 tag, two differences stand out. Trunk has a new rebalance-listener API (RebalanceListener, KafkaConsumer.setRebalanceListener()) that 4.4.0 does not have. And 4.4.0's SubscriptionState class has been split in trunk into ConsumerSubscriptionState and a shared interface, AbstractSubscriptionState (which the share consumer's ShareSubscriptionState also implements). The rest of the quoted code is the same as 4.4.0 or differs by a few lines.

On the consumer side alone, AsyncKafkaConsumer.java is 2,557 lines and ConsumerCoordinator.java 1,763 lines, and the broker's GroupMetadataManager.java is 10,188 lines. Those files also hold code for Kafka Streams groups and share groups. This post picks only the main path a newcomer can follow; what was left out is listed by name and location in "What this post leaves out" at the end.

Terms to know first

Consumer group and group coordinator: consumers that use the same group.id form a consumer group, and each partition of a topic goes to only one consumer in the group. Deciding and remembering who gets which partition is the job of one of the brokers; that role (more precisely, a component inside the broker) is called the group coordinator. Each group has its own broker in charge, and that broker is the leader of the __consumer_offsets partition chosen from the group id (section 5).

Rebalance: redistributing partitions when the group's members change (join, leave, stop responding) or the subscribed topics' partition count changes. In your app you see it through the onPartitionsRevoked (give up) and onPartitionsAssigned (receive) callbacks of a ConsumerRebalanceListener.

group.protocol: a consumer setting that picks how rebalances are carried out. classic is the long-standing way: it uses JoinGroup, SyncGroup and Heartbeat requests, and one of the consumers (the leader) computes the assignment. consumer is the new way introduced by KIP-848: it uses only the ConsumerGroupHeartbeat request, and the broker computes the assignment (the javadoc of GroupProtocol.java calls the two values "pre KIP-848" and "KIP-848"). At this commit the default is still classic (section 1).

Generation and epoch: numbers that go up each time a rebalance finishes. In classic, the whole group has one generation. In the consumer protocol there are separate numbers: the group epoch (goes up when the group's makeup changes), the target assignment epoch, and the member epoch (how far this member has caught up). Requests carrying an old number are rejected — this stops a "zombie" consumer from committing offsets late for partitions it has already given up.

Assignor: a class that takes a list of members and a list of partitions and computes "who gets what". In classic it is the client's partition.assignment.strategy (RangeAssignor, CooperativeStickyAssignor …); in the consumer protocol it is the broker's group.consumer.assignors (uniform, range).

Eager and cooperative: two ways of giving up partitions during a rebalance. Eager means every member gives up all its partitions when a rebalance starts and then receives a new assignment. Cooperative means a member gives up only the partitions that actually move to another member; the rest can still be read during the rebalance.

__consumer_offsets: the internal topic the group coordinator uses. Despite the name, it holds not only committed offsets but also group state such as members and assignments — all as records (section 6). It has 50 partitions by default (offsets.topic.num.partitions, lines 120–121 of GroupCoordinatorConfig.java).

A few terms from earlier posts come back. The high watermark (HW) is the offset boundary up to which records are replicated to every ISR replica and count as "committed" (post 9). The request handler threads (data-plane-kafka-request-handler-N) and the match in KafkaApis.handle() were covered in post 4.

1. KafkaConsumer is a shell — two implementations split by group.protocol

What gets created when your app calls new KafkaConsumer<>(props)? Open KafkaConsumer and it has essentially one field.

    private static final ConsumerDelegateCreator CREATOR = new ConsumerDelegateCreator();

    private final ConsumerDelegate<K, V> delegate;
    // ...
    KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
        delegate = CREATOR.create(config, keyDeserializer, valueDeserializer);
    }

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java lines 554–629

Public methods like poll(), subscribe() and commitSync() all forward in one line, like delegate.poll(timeout) (for example, poll() is lines 986–989). Most of the 2,027-line file is javadoc. The real implementation is chosen in ConsumerDelegateCreator.

    public <K, V> ConsumerDelegate<K, V> create(ConsumerConfig config,
                                                Deserializer<K> keyDeserializer,
                                                Deserializer<V> valueDeserializer) {
        try {
            GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

            if (groupProtocol == GroupProtocol.CONSUMER)
                return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty());
            else
                return new ClassicKafkaConsumer<>(config, keyDeserializer, valueDeserializer);
        } catch (KafkaException e) {
            throw e;
        } catch (Throwable t) {
            throw new KafkaException("Failed to construct Kafka consumer", t);
        }
    }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java lines 57–72

group.protocol=consumer gives you AsyncKafkaConsumer; anything else gives ClassicKafkaConsumer. The default is in ConsumerConfig.

    public static final String GROUP_PROTOCOL_CONFIG = "group.protocol";
    public static final String DEFAULT_GROUP_PROTOCOL = GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT);
    public static final String GROUP_PROTOCOL_DOC = "The group protocol that the consumer uses. The " +
        "supported values are <code>classic</code> or <code>consumer</code>. The default value is <code>classic</code>.";

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java lines 120–123

define(GROUP_PROTOCOL_CONFIG, Type.STRING, DEFAULT_GROUP_PROTOCOL, …) (lines 684–689) uses this constant as the default. So an app that doesn't set it still uses ClassicKafkaConsumer today. 4.4.0 is the same.

The two implementations already differ in how they split work between threads. Blue boxes are threads; yellow boxes are the queues between them.

Easy to get wrong: some settings are forbidden depending on the protocol

The two implementations also accept slightly different settings. With group.protocol=consumer, ConsumerConfig throws if you set partition.assignment.strategy, heartbeat.interval.ms or session.timeout.ms (the two lists at lines 423–438 and checkUnsupportedConfigsPostProcess() at lines 819–842). In the new protocol these three moved to broker-side settings (group.consumer.assignors, group.consumer.heartbeat.interval.ms, group.consumer.session.timeout.ms). The other way round, with classic you can't set group.remote.assignor (the setting that picks a broker-side assignor by name). If your existing app's config contains session.timeout.ms, changing only group.protocol won't even let the consumer be created.

2. The classic consumer's poll() — one thread does almost everything

Let's start with the default, ClassicKafkaConsumer. poll(Duration) creates a timer and runs this loop.

            do {
                client.maybeTriggerWakeup();

                // try to update assignment metadata BUT do not need to block on the timer for join group
                updateAssignmentMetadataIfNeeded(timer, false);

                final Fetch<K, V> fetch = pollForFetches(timer);
                if (!fetch.isEmpty()) {
                    // ...
                    if (sendFetches() > 0 || client.hasPendingRequests()) {
                        client.transmitSends();
                    }
                    // ...
                    return this.interceptors.onConsume(new ConsumerRecords<>(fetch.records(), fetch.nextOffsets()));
                }
            } while (timer.notExpired());

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java lines 673–698

One pass of the loop has two steps.

1. updateAssignmentMetadataIfNeeded() — the group work. It calls coordinator.poll() (ConsumerCoordinator) and then decides the starting position of any partition that doesn't have one yet (the committed offset or auto.offset.reset) (lines 712–718).
2. pollForFetches() — getting records. If some are already buffered it returns them right away; otherwise it sends Fetch requests (sendFetches()) and waits for responses in the network client's poll() (lines 723–756). The inside of Fetch is post 8's job.

Also worth noting: right before returning records it sends the next Fetch ahead of time (the comment at lines 681–689), so the next batch is on its way while your app processes the current one.

The heart of the group work, ConsumerCoordinator.poll(), looks like this.

    public boolean poll(Timer timer, boolean waitForJoinGroup) {
        maybeUpdateSubscriptionMetadata();

        invokeCompletedOffsetCommitCallbacks();

        if (subscriptions.hasAutoAssignedPartitions()) {
            // ...
            pollHeartbeat(timer.currentTimeMs());
            if (coordinatorUnknownAndUnreadySync(timer)) {
                return false;
            }

            if (rejoinNeededOrPending()) {
                // ...
                // if not wait for join group, we would just use a timer of 0
                if (!ensureActiveGroup(waitForJoinGroup ? timer : time.timer(0L))) {
                    // ...
                    return false;
                }
            }
        } else {
            // ...
        }

        maybeAutoCommitOffsetsAsync(timer.currentTimeMs());
        return true;
    }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java lines 513–579

If you subscribed with subscribe() (hasAutoAssignedPartitions()), it does the following in order.

1. pollHeartbeat() — records the time "the app just called poll()". The heartbeat thread below measures max.poll.interval.ms from this time.
2. coordinatorUnknownAndUnreadySync() — if it doesn't know which broker is the group coordinator, it finds out with a FindCoordinator request (AbstractCoordinator.lookupCoordinator(), line 331).
3. If rejoinNeededOrPending(), it (re)joins the group with ensureActiveGroup(). The rebalance happens right here, inside the poll() your app called. That is also why rebalance callbacks run on the thread that called poll().
4. maybeAutoCommitOffsetsAsync() — if enable.auto.commit=true (the default, ConsumerConfig.java lines 489–493), it commits offsets asynchronously every auto.commit.interval.ms (default 5000, lines 494–499). Auto-commit has no thread of its own; each poll() just checks whether it's time.

ensureActiveGroup() (AbstractCoordinator.java lines 415–424) checks the coordinator, starts the heartbeat thread with startHeartbeatThreadIfNeeded(), and joins the group with joinGroupIfNeeded(). The heartbeat thread is created once, the first time the consumer joins. Its name is "kafka-coordinator-heartbeat-thread" + " | " + group.id (the constant at line 124 and line 1467). This thread only sends Heartbeat requests; the rebalance itself (JoinGroup and SyncGroup) is done by the app thread in joinGroupIfNeeded().

The two clocks the heartbeat thread watches

The key lines of the heartbeat thread's run() loop are these.

                        } else if (heartbeat.sessionTimeoutExpired(now)) {
                            // the session timeout has expired without seeing a successful heartbeat, so we should
                            // probably make sure the coordinator is still healthy.
                            markCoordinatorUnknown("session timed out without receiving a "
                                    + "heartbeat response");
                        } else if (heartbeat.pollTimeoutExpired(now)) {
                            // the poll timeout has expired, which means that the foreground thread has stalled
                            // in between calls to poll().
                            handlePollTimeoutExpiry();

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java lines 1523–1531

There are two clocks.

• Session clock (session.timeout.ms, default 45000): if no heartbeat response arrives for this long, the client assumes something is wrong with the coordinator and looks it up again. The broker, too, removes the member from the group if it hears no heartbeat for that long.
• Poll clock (max.poll.interval.ms, default 300000): if your app doesn't call poll() for this long, the heartbeat thread leaves the group on its own (handlePollTimeoutExpiry()). Because heartbeats run separately, without this clock an app stuck processing records could hold its partitions forever.

The heartbeat interval is heartbeat.interval.ms (default 3000). I checked all three defaults in the define(…) calls at lines 467–476 and 658–663 of ConsumerConfig.java. And max.poll.interval.ms also serves as the rebalance timeout when joining the group (GroupRebalanceConfig.java line 55). That value comes back in the next section.

Rebalances and auto commit happen inside the poll() the application calls; the two clocks live in the heartbeat thread. The poll clock is measured from the time the application thread noted. Numbers in parentheses are defaults.

3. The classic rebalance — JoinGroup, SyncGroup and the leader consumer

A classic rebalance is two round trips.

1. JoinGroup: each member says "I'm joining this group, I subscribe to these topics, and I can use these assignors" (sendJoinGroupRequest(), AbstractCoordinator.java lines 608–638). The coordinator holds the responses until all members have come in, then picks one as the leader and gives only the leader everyone's subscriptions.
2. SyncGroup: the leader computes the assignment inside the client from those subscriptions and sends it in SyncGroup; the other members send an empty SyncGroup (onJoinFollower(), lines 755–770). The coordinator splits the leader's assignment per member and returns it in the SyncGroup responses.

When JoinGroupResponseHandler gets the JoinGroup response, it branches on joinResponse.isLeader(): the leader goes to onLeaderElected(), everyone else to onJoinFollower() (lines 678–682). On the leader, ConsumerCoordinator.onLeaderElected() unpacks the members' subscriptions and computes the assignment with assignor.assign(metadata.fetch(), new GroupSubscription(subscriptions)) (line 693). The broker never interprets the assignment; it just passes the bytes along. That is why, in classic, changing the assignment strategy or using a new assignor meant redeploying every consumer app.

The default is eager

The default partition.assignment.strategy is two assignors, [RangeAssignor, CooperativeStickyAssignor] (ConsumerConfig.java lines 477–482). But the rebalance protocol is chosen as the most advanced one that every assignor supports.

        if (!assignors.isEmpty()) {
            List<RebalanceProtocol> supportedProtocols = new ArrayList<>(assignors.get(0).supportedProtocols());

            for (ConsumerPartitionAssignor assignor : assignors) {
                supportedProtocols.retainAll(assignor.supportedProtocols());
            }

            // ...
            Collections.sort(supportedProtocols);

            protocol = supportedProtocols.get(supportedProtocols.size() - 1);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java lines 256–271

RangeAssignor supports only EAGER, so after the intersection (retainAll) the default rebalance protocol is EAGER. Having CooperativeStickyAssignor in the list leaves a path open to drop Range later and move to cooperative; it does not mean you get cooperative behaviour today. Here is what EAGER does right before a rebalance (onJoinPrepare()).

            switch (protocol) {
                case EAGER:
                    // revoke all partitions
                    revokedPartitions.addAll(subscriptions.assignedPartitions());
                    exception = rebalanceListenerInvoker.invokePartitionsRevoked(revokedPartitions);

                    subscriptions.assignFromSubscribed(Collections.emptySet());

                    break;

                case COOPERATIVE:
                    // only revoke those partitions that are not in the subscription anymore.
                    Set<TopicPartition> ownedPartitions = new HashSet<>(subscriptions.assignedPartitions());
                    revokedPartitions.addAll(ownedPartitions.stream()
                        .filter(tp -> !subscriptions.subscription().contains(tp.topic()))
                        .collect(Collectors.toSet()));

                    if (!revokedPartitions.isEmpty()) {
                        exception = rebalanceListenerInvoker.invokePartitionsRevoked(revokedPartitions);

                        ownedPartitions.removeAll(revokedPartitions);
                        subscriptions.assignFromSubscribed(ownedPartitions);
                    }

                    break;
            }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java lines 828–853

EAGER gives up all its partitions (all of them go to onPartitionsRevoked) and sends JoinGroup with an empty assignment. Until it gets a new assignment, this consumer reads nothing.

B joins while consumer A holds all six partitions alone. This is the same order as the run of example 1 below. In reality the new member B's first JoinGroup comes back with a MEMBER_ID_REQUIRED error carrying only a member id, and B immediately resends with that id (GroupMetadataManager.classicGroupJoinNewDynamicMember(), lines 7416–7437); the diagram folds that round trip into step 1.

Steps 3–4 matter. The coordinator has no way to reach out to A first and say "let's rebalance". It can only put REBALANCE_IN_PROGRESS in A's next heartbeat response (HeartbeatResponseHandler, requestRejoin() at lines 1257–1268), and A sends JoinGroup on the poll() after that. So it takes up to about heartbeat.interval.ms (3 s) after B joins before A gives up its partitions.

Group states on the coordinator side

On the broker, a classic group moves between five states. The javadoc of ClassicGroupState describes, almost like a table, how each state answers each request, and the allowed previous states are in a static block.

    static {
        EMPTY.addValidPreviousStates(PREPARING_REBALANCE);
        PREPARING_REBALANCE.addValidPreviousStates(STABLE, COMPLETING_REBALANCE, EMPTY);
        COMPLETING_REBALANCE.addValidPreviousStates(PREPARING_REBALANCE);
        STABLE.addValidPreviousStates(COMPLETING_REBALANCE);
        DEAD.addValidPreviousStates(STABLE, PREPARING_REBALANCE, COMPLETING_REBALANCE, EMPTY, DEAD);
    }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroupState.java lines 109–115

The two yellow states are "rebalancing". Besides the one arrow shown, a group can go to Dead from any state when its __consumer_offsets partition moves to another broker.

How long PreparingRebalance waits is decided by GroupMetadataManager.prepareRebalance() (lines 7952–7985). When the first member joins an empty group, the coordinator waits group.initial.rebalance.delay.ms (default 3000, GroupCoordinatorConfig.java line 180) for more members, so that several apps starting at nearly the same time don't trigger rebalance after rebalance. For a group that already has members, it waits until all existing members have sent JoinGroup again, at most the group's rebalance timeout (the largest max.poll.interval.ms among the members, ClassicGroup.rebalanceTimeoutMs() lines 767–773) (maybeCompleteJoinElseSchedule(), lines 7994–8011). The examples in this post start the server with group.initial.rebalance.delay.ms=0, set by the helper script.

Example 1: adding two consumers one after the other (classic)

This example was run with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411), in a Docker container (Eclipse Temurin JDK 25). We put 60 records into a six-partition topic orders, then start consumer A in one JVM. Once A has all six, we start B; once the two have three each, we wait 20 seconds and close B. Each consumer runs poll(100ms) on its own thread (app-A, app-B) and prints when, and on which thread, its rebalance callbacks run.

// This example was written for this post (it is not Kafka source)
// ...
                consumer.subscribe(List.of(topic), new ConsumerRebalanceListener() {
                    public void onPartitionsRevoked(Collection<TopicPartition> p) {
                        log(name, "onPartitionsRevoked  " + parts(p) + "  thread=" + Thread.currentThread().getName());
                    }
                    // ...
                });
                while (running.get()) {
                    consumed.addAndGet(consumer.poll(Duration.ofMillis(100)).count());
                }

The whole program is a single Rebalance.java that takes group.protocol and the group id as arguments. Here is the run with group.protocol=classic (times and member ids differ on every run).

group.protocol=classic  group.id=g-classic  -> __consumer_offsets-18
  1013 ms [main] start A
  3089 ms [A] onPartitionsAssigned [0, 1, 2, 3, 4, 5]  thread=app-A  now owns [0, 1, 2, 3, 4, 5]
  3118 ms [main] A is stable. client threads:
         thread: app-A
         thread: kafka-coordinator-heartbeat-thread | g-classic
  3121 ms [main] start B
  6116 ms [A] onPartitionsRevoked  [0, 1, 2, 3, 4, 5]  thread=app-A
  6147 ms [A] onPartitionsAssigned [0, 1, 2]  thread=app-A  now owns [0, 1, 2]
  6148 ms [B] onPartitionsAssigned [3, 4, 5]  thread=app-B  now owns [3, 4, 5]
  6167 ms [main] A and B are stable. client threads:
         thread: app-A
         thread: app-B
         thread: kafka-coordinator-heartbeat-thread | g-classic
 24451 ms [main] stop B
 24501 ms [B] close() (consumed 0 records)
 24513 ms [B] onPartitionsRevoked  [3, 4, 5]  thread=app-B
 24706 ms [B] closed
 27154 ms [A] onPartitionsRevoked  [0, 1, 2]  thread=app-A
 27161 ms [A] onPartitionsAssigned [0, 1, 2, 3, 4, 5]  thread=app-A  now owns [0, 1, 2, 3, 4, 5]
 27195 ms [main] stop A
 27288 ms [A] close() (consumed 60 records)
 27289 ms [A] onPartitionsRevoked  [0, 1, 2, 3, 4, 5]  thread=app-A
 27399 ms [A] closed
 27401 ms [main] records consumed: A=60 B=0

Three things to read here.

• When B joins, A gives up all six (onPartitionsRevoked [0, 1, 2, 3, 4, 5]) and then gets three back. That's because the default is EAGER. Even 0, 1 and 2, which A keeps, are briefly let go.
• About 3 seconds pass between starting B and A giving up its partitions. As we saw, A only learns about the rebalance from its next heartbeat response (every 3 s). The same happens when B closes: B sends LeaveGroup as it closes, but A notices on its next heartbeat, gives up everything again, and gets all six.
• The callbacks always run on app-A or app-B, that is, on the thread that called poll(). Besides those, the list shows kafka-coordinator-heartbeat-thread | g-classic. It appears only once even after B joins because A's and B's heartbeat threads use the same name (same group id) and the example collects names into a TreeSet — there are really two, one per consumer. (The list is filtered by client-ish names; it shows there is no other client thread such as kafka-producer-network-thread, not every thread in the JVM.)

The time until A first got all six partitions varied from 2 to 8 seconds across the four runs I made (about 2 seconds in the output above). This group is the first on the broker to use __consumer_offsets, so the first FindCoordinator makes the broker create that internal topic (50 partitions), and until the topic is ready the client gets COORDINATOR_NOT_AVAILABLE (KafkaApis.scala lines 1292–1295). I believe that wait explains the variation, but I did not break the time down to confirm it.

4. The consumer protocol — a background thread and a single heartbeat

Now AsyncKafkaConsumer, which you get with group.protocol=consumer. As the "Async" in the name suggests, the app thread never touches the network directly. The constructor starts one ConsumerNetworkThread, named consumer_background_thread (ConsumerNetworkThread.java line 68). The poll() loop looks like classic's, but does different work.

        if (inflightPoll == null) {
            inflightPoll = new AsyncPollEvent(calculateDeadlineMs(timer), time.milliseconds());
            newlySubmittedEvent = true;
            log.trace("Inflight event {} submitted", inflightPoll);
            applicationEventHandler.add(inflightPoll);
        }

        try {
            // Note: this is calling user-supplied code, so make sure that any errors thrown here are caught and
            // the inflight event is cleared.
            offsetCommitCallbackInvoker.executeCallbacks();
            processBackgroundEvents();

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java lines 1006–1017

The app thread puts an AsyncPollEvent into the event queue (applicationEventQueue) to tell the background thread "poll was called", and handles events sent back by the background thread (backgroundEventQueue) in processBackgroundEvents(). Then, in pollForFetches(), it waits on the FetchBuffer that the background thread fills (fetchBuffer.awaitWakeup(pollTimer), line 2026). Turning the received bytes into records is done on the app thread by FetchCollector.

One pass of the background thread looks like this.

        long pollWaitTimeMs = MAX_POLL_TIMEOUT_MS;

        for (RequestManager rm : requestManagers.entries()) {
            NetworkClientDelegate.PollResult pollResult = rm.poll(currentTimeMs);
            long timeoutMs = networkClientDelegate.addAll(pollResult);
            pollWaitTimeMs = Math.min(pollWaitTimeMs, timeoutMs);
        }

        networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java lines 220–228

runOnce() (lines 210–242) first processes the events the app thread queued (processApplicationEvents()), then asks each request manager in turn "do you have a request to send now?" (rm.poll()), and hands the collected requests to the network client, which sends them and receives responses. The list of request managers is in the RequestManagers constructor — coordinator lookup (CoordinatorRequestManager), offset commits (CommitRequestManager), heartbeats (ConsumerHeartbeatRequestManager), membership (ConsumerMembershipManager), offset lookups, topic metadata and Fetch (FetchRequestManager), in that order (RequestManagers.java lines 94–105). Each kind of request has its own class, and a single thread calls them in rotation. Much like the broker's Processor in post 4 looked after many connections with one selector, here one thread looks after many kinds of requests.

The heartbeat is the rebalance

In the new protocol, practically the only request the consumer exchanges with the group is ConsumerGroupHeartbeat (API key 68). It is used to join, to receive assignments, to report that partitions were given up, and to leave. What goes into the heartbeat tells you how the protocol works.

            // MemberEpoch - always sent
            data.setMemberEpoch(membershipManager.memberEpoch());

            // ...
            boolean sendAllFields = membershipManager.state() == MemberState.JOINING;

            // RebalanceTimeoutMs - only sent when joining or if it has changed since the last heartbeat
            if (sendAllFields || sentFields.rebalanceTimeoutMs != rebalanceTimeoutMs) {
                data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
                sentFields.rebalanceTimeoutMs = rebalanceTimeoutMs;
            }

            // SubscribedTopicNames - only sent if it has changed since the last heartbeat
            TreeSet<String> subscribedTopicNames = new TreeSet<>(this.subscriptions.subscription());
            if (sendAllFields || !subscribedTopicNames.equals(sentFields.subscribedTopicNames)) {
                data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
                sentFields.subscribedTopicNames = subscribedTopicNames;
            }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java lines 266–285

The group id, member id and member epoch go out every time; the rest — subscribed topics, rebalance timeout, the partitions it currently owns (TopicPartitions, lines 304–314) — are sent only when they change (sentFields remembers the last values). When joining (JOINING), everything is sent. Unlike classic, the member id is generated by the client up front (the comment at line 263, and Uuid.randomUuid() at line 82 of AbstractMembershipManager.java). The heartbeat interval also comes not from a client setting but from the value the broker puts in the response (group.consumer.heartbeat.interval.ms, default 5000).

When a response brings a new assignment, AbstractMembershipManager.maybeReconcile() compares it with what the consumer owns.

        // Partitions to assign (not previously owned)
        SortedSet<TopicPartition> addedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
        addedPartitions.addAll(assignedTopicPartitions);
        addedPartitions.removeAll(ownedPartitions);

        // Partitions to revoke
        SortedSet<TopicPartition> revokedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
        revokedPartitions.addAll(ownedPartitions);
        revokedPartitions.removeAll(assignedTopicPartitions);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java lines 916–924

"To add = new assignment − owned" and "to revoke = owned − new assignment". It then goes commit (if auto-commit is on) → onPartitionsRevoked → onPartitionsAssigned (lines 956–970, revokeAndAssign() lines 994–1036), and when done switches to ACKNOWLEDGING so that the next heartbeat goes out right away, without waiting for the interval, to say "done" (line 1031). Only the partitions being given up reach the callback, so this protocol is cooperative from the start.

The background thread doesn't call the callbacks itself.

    private CompletableFuture<Void> enqueueConsumerRebalanceListenerCallback(ConsumerRebalanceListenerMethodName methodName,
                                                                             Set<TopicPartition> partitions) {
        SortedSet<TopicPartition> sortedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
        sortedPartitions.addAll(partitions);

        CompletableBackgroundEvent<Void> event = new PartitionsRemovedEvent(methodName, sortedPartitions);
        backgroundEventHandler.add(event);
        log.debug("The event to trigger the {} method execution was enqueued successfully", methodName.fullyQualifiedMethodName());
        return event.future();
    }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java lines 415–424

The background thread puts a "please run this callback" event into the app-side queue; the app thread runs the callback in processBackgroundEvents() during its next poll() and completes the future. The background thread's rebalance handling waits until that future completes. You need to be able to call consumer methods such as commitSync() inside a callback, and KafkaConsumer is meant to be used from a single thread, so the callbacks run on the app thread.

The background thread does all the networking, but the rebalance callback, being user code, is handed to the application thread as an event and run there.

max.poll.interval.ms is now measured by the background thread

The poll clock that classic's heartbeat thread kept is measured here by the heartbeat request manager (AbstractHeartbeatRequestManager.poll(), lines 171–186). If poll() isn't called for too long, it sends a heartbeat that leaves the group. The "Consumer poll timeout has expired" warning you often see in app logs comes from here (line 173). The client-side member states (MemberState — JOINING, RECONCILING, ACKNOWLEDGING, STABLE, FENCED, PREPARE_LEAVING, LEAVING, STALE and others, 10 in total) are defined in MemberState.java, and the allowed transitions are in the static block at lines 118–142. When the poll clock runs out, the member goes through LEAVING to STALE, and on the next poll() it rejoins through JOINING.

Example 1, continued: the same thing with the consumer protocol

Here is the same Rebalance.java run with group.protocol=consumer.

group.protocol=consumer  group.id=g-consumer  -> __consumer_offsets-32
  1097 ms [main] start A
  2741 ms [A] onPartitionsAssigned [0, 1, 2, 3, 4, 5]  thread=app-A  now owns [0, 1, 2, 3, 4, 5]
  2781 ms [main] A is stable. client threads:
         thread: app-A
         thread: consumer_background_thread
  2784 ms [main] start B
  3037 ms [B] onPartitionsAssigned []  thread=app-B  now owns []
  7957 ms [A] onPartitionsRevoked  [3, 4, 5]  thread=app-A
  8040 ms [A] onPartitionsAssigned []  thread=app-A  now owns [0, 1, 2]
 13102 ms [B] onPartitionsAssigned [3, 4, 5]  thread=app-B  now owns [3, 4, 5]
 13147 ms [main] A and B are stable. client threads:
         thread: app-A
         thread: app-B
         thread: consumer_background_thread
 31443 ms [main] stop B
 31526 ms [B] close() (consumed 0 records)
 31536 ms [B] onPartitionsRevoked  [3, 4, 5]  thread=app-B
 31908 ms [B] closed
 33084 ms [A] onPartitionsAssigned [3, 4, 5]  thread=app-A  now owns [0, 1, 2, 3, 4, 5]
 33093 ms [main] stop A
 33161 ms [A] close() (consumed 60 records)
 33167 ms [A] onPartitionsRevoked  [0, 1, 2, 3, 4, 5]  thread=app-A
 33598 ms [A] closed
 33600 ms [main] records consumed: A=60 B=0

Put next to classic, the differences are clear.

• When B joins, A gives up only 3, 4 and 5 (onPartitionsRevoked [3, 4, 5]). Partitions 0, 1 and 2 stay A's throughout the rebalance. A's onPartitionsAssigned [] means "nothing new"; it is always called once after revoking finishes (the javadoc of revokeAndAssign(), lines 988–993).
• B gets an empty assignment ([]) right after joining, and [3, 4, 5] only much later. The broker won't give B those partitions until A reports that it has let go of 3, 4 and 5 (section 5).
• About 5 seconds pass between starting B and A giving up its partitions — exactly the heartbeat interval the broker handed out (5000 ms). A only learns about the new assignment on its next heartbeat.
• B needed about 10 seconds — two heartbeat intervals — to get [3, 4, 5]. In this run, B's regular heartbeat reached the broker a few milliseconds before A's "I've let go" heartbeat, so at that moment there was nothing for B yet, and B only got them on its following heartbeat 5 seconds later. The records in example 3 show that exact order. In earlier runs it took about 5.5 seconds — it depends on the order of those two heartbeats.
• When B leaves, A gives up nothing and simply gets [3, 4, 5] added.
• Instead of kafka-coordinator-heartbeat-thread there is a consumer_background_thread. As with classic's heartbeat thread, there are two consumers but only one name in the list, because both consumers' background threads use the same name (super(BACKGROUND_THREAD_NAME, true), line 96 — no client id is added). The example collects names into a TreeSet, so a repeated name shows up once. When you look at an app with jstack, remember that each consumer has one such thread.

Easy to get wrong: the new listener API and the old one

The example uses subscribe(topics, ConsumerRebalanceListener), which app developers know well. At this commit (trunk), the KafkaConsumer javadoc calls this method "deprecated" and suggests calling subscribe(topics) and setRebalanceListener(RebalanceListener) separately (lines 713–715 and 885–915). Callbacks of the new RebalanceListener get a RebalanceConsumer (a consumer view that exposes only methods safe to use during a rebalance) as a second argument. However, the method carries no @Deprecated annotation, so there is no compiler warning, and 4.4.0 has no RebalanceListener at all. If your code must run on 4.4.0, keep using ConsumerRebalanceListener as in the example.

5. On the broker: where the group coordinator lives and how it decides assignments

Now we cross over to the side that receives these requests. The group coordinator is not a separate server; it is a component inside every broker, and only the broker in charge differs from group to group. A client first sends FindCoordinator to any broker to ask which one is in charge.

      val (partition, internalTopicName) = CoordinatorType.forId(keyType) match {
        case CoordinatorType.GROUP =>
          (groupCoordinator.partitionFor(key), GROUP_METADATA_TOPIC_NAME)

        case CoordinatorType.TRANSACTION =>
          (txnCoordinator.partitionFor(key), TRANSACTION_STATE_TOPIC_NAME)

        case CoordinatorType.SHARE =>
          // We know that shareCoordinator is defined at this stage.
          (shareCoordinator.partitionFor(SharePartitionKey.getInstance(key)), SHARE_GROUP_STATE_TOPIC_NAME)
      }

core/src/main/scala/kafka/server/KafkaApis.scala lines 1277–1287

Reading Scala — x match { case A => … case B => … } is like a Java switch expression, and the last expression of the chosen branch becomes the value. Here each branch returns a tuple (partition number, topic name) (two values bundled together), and val (partition, internalTopicName) = … unpacks that tuple into two variables.

For a group, it gets the __consumer_offsets partition number from groupCoordinator.partitionFor(group id) and returns that partition's leader broker as the coordinator (lines 1300–1307). The calculation is simple.

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

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java lines 455–460

It's the group id's hashCode() modulo the partition count (50 by default). In the example, g-classic landed on partition 18 and g-consumer on partition 32 (the example computed and printed them with the same formula). This spreads groups evenly over the 50 partitions, and so over the brokers that lead them. And because a group's state is written to that partition, when the leader changes the coordinator naturally moves to the new leader. That comes for free from the replication of post 9.

How one request reaches the coordinator

Let's follow ConsumerGroupHeartbeat. KafkaApis.handleConsumerGroupHeartbeat() (lines 2680–2718) checks authorization, calls groupCoordinator.consumerGroupHeartbeat(…), and sends the response when the returned CompletableFuture completes. GroupCoordinatorService schedules it as a write operation.

        return runtime.scheduleWriteOperation(
            "consumer-group-heartbeat",
            topicPartitionFor(request.groupId()),
            coordinator -> coordinator.consumerGroupHeartbeat(context, request)
        ).exceptionally(exception -> handleOperationException(

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java lines 535–539

A "heartbeat" being a write may look odd at first. It's because a heartbeat can change group state (members, assignments, epochs), and changed state has to be written to __consumer_offsets. What to change is decided by GroupMetadataManager.consumerGroupHeartbeat(). An epoch of -1 (or -2 for a static member) branches off as a request to leave the group (lines 5729–5737); everything else goes through three steps, which the comments in the method number for us. Step 1 creates or updates the member and, if its subscription changed, bumps the group epoch (lines 2635–2703). Steps 2 and 3 look like this.

        // 2. Update the target assignment if the group epoch is larger than the target assignment epoch. The delta between
        // the existing and the new target assignment is persisted to the partition.
        UpdateTargetAssignmentResult<Assignment> updateTargetAssignmentResult = maybeUpdateTargetAssignment(
            group,
            groupEpoch,
            member,
            updatedMember,
            subscriptionType,
            records
        );

        // 3. Reconcile the member's assignment with the target assignment if the member is not
        // fully reconciled yet.
        updatedMember = maybeReconcile(
            groupId,
            updatedMember,
            group::currentPartitionEpoch,
            updateTargetAssignmentResult.targetAssignmentEpoch(),
            updateTargetAssignmentResult.targetAssignment(),
            group.resolvedRegularExpressions(),
            // Force consistency with the subscription when the subscription has changed.
            hasSubscriptionChanged,
            ownedTopicPartitions,
            records
        );

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java lines 2705–2729

2. Target assignment: if the group epoch is larger than the target assignment epoch, the broker's assignor (uniform by default, GroupCoordinatorConfig.java lines 221–233) computes a new target assignment for the whole group (maybeUpdateTargetAssignment(), lines 4346–4429). The job of computing assignments has moved from the client leader to the broker. But if group.consumer.assignment.interval.ms (default 1000) hasn't passed since the previous computation, it skips the computation this time and leaves it to the next heartbeat that comes in (from any member) (lines 4359–4366, canComputeNextTargetAssignment() lines 4312–4333). This keeps the broker from recomputing assignments over and over when members join in a burst.
3. Reconcile: the target is the picture for the whole group; what this member can hold right now is computed separately, by CurrentAssignmentBuilder inside maybeReconcile().

All records go into the records list and are returned together with the response as a CoordinatorResult at the end (line 2750). The assignment is put into the response only when the member just joined, the request carried all fields, or this member's assignment changed (lines 2739–2748).

Between target and current: CurrentAssignmentBuilder

Even if the target becomes A=[0,1,2], B=[3,4,5], giving 3, 4 and 5 to B right away would let A and B read the same partitions at the same time, because A hasn't yet reported letting go of them. CurrentAssignmentBuilder.computeNextAssignment() closes that gap.

            // New Assigned Partitions = Previous Assigned Partitions ∩ Target
            Map<Integer, Integer> assignedPartitions = new HashMap<>(currentAssignedPartitions);
            assignedPartitions.keySet().retainAll(target);

            // Partitions Pending Revocation = Previous Assigned Partitions - New Assigned Partitions
            Map<Integer, Integer> partitionsPendingRevocation = new HashMap<>(currentAssignedPartitions);
            partitionsPendingRevocation.keySet().removeAll(assignedPartitions.keySet());

            // Partitions Pending Assignment = Target - New Assigned Partitions - Unreleased Partitions
            Set<Integer> partitionsPendingAssignment = new HashSet<>(target);
            partitionsPendingAssignment.removeAll(assignedPartitions.keySet());
            hasUnreleasedPartitions = partitionsPendingAssignment.removeIf(partitionId ->
                currentPartitionEpoch.apply(topicId, partitionId) != -1 &&
                // Don't consider a partition unreleased if it is owned by the current member
                // because it is pending revocation. This is safe to do since only a single member
                // can own a partition at a time.
                !member.partitionsPendingRevocation().getOrDefault(topicId, Map.of()).containsKey(partitionId)
            ) || hasUnreleasedPartitions;

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java lines 393–410

The comments are set formulas.

• Keep = currently owned ∩ target
• Give up = currently owned − keep
• Newly receive = target − keep − partitions still held by another member (currentPartitionEpoch(…) != -1)

If there is something to give up, the member can't move to the new epoch and goes to UNREVOKED_PARTITIONS (lines 425–435). If something it should receive is still held by another member, it moves to the new epoch but waits in UNRELEASED_PARTITIONS (lines 436–465). If neither, it is STABLE. These three states (plus UNKNOWN, reserved for the future) are the broker-side member states defined in modern/MemberState.java — a different enum with the same name as the client-side MemberState of section 4, so be careful not to mix them up.

The switch in CurrentAssignmentBuilder.build() (lines 187–260) as a picture. Rare arrows, such as the target changing again while in one of the two waiting (yellow) states, are left out.

Thanks to this, a partition is owned by only one member at any moment, while partitions that don't move keep being read. It gets the safety that classic's EAGER got by "everyone puts everything down and picks it up again", but per partition.

The consumer-protocol run of example 1 (the output above) drawn request by request, matched with the records of example 3. The order of steps 8–11 can change from run to run. The one onPartitionsAssigned [] call on B after step 3 is left out of the picture.

Example 2: the two groups through kafka-consumer-groups.sh

While example 1 was running, once A and B had settled, we called kafka-consumer-groups.sh --describe --state and --describe --members --verbose in the same container.

=== 1. group.protocol=classic
--- kafka-consumer-groups.sh --describe --state (while A and B are stable)

GROUP           COORDINATOR (ID)          ASSIGNMENT-STRATEGY  STATE                #MEMBERS
g-classic       localhost:9092  (1)       range                Stable               2
--- kafka-consumer-groups.sh --describe --members

GROUP           CONSUMER-ID                                      HOST            CLIENT-ID       #PARTITIONS     CURRENT-EPOCH   CURRENT-ASSIGNMENT   TARGET-EPOCH    TARGET-ASSIGNMENT   
g-classic       g-classic-A-386cac32-ece8-4c7b-a804-fec4152c1e6e /127.0.0.1      g-classic-A     3               -               orders:0,1,2         -               -                   
g-classic       g-classic-B-e2fd2ef9-937d-4eb2-8c87-370a1089cb72 /127.0.0.1      g-classic-B     3               -               orders:3,4,5         -               -
=== 2. group.protocol=consumer
--- kafka-consumer-groups.sh --describe --state (while A and B are stable)

GROUP           COORDINATOR (ID)          ASSIGNMENT-STRATEGY  STATE                #MEMBERS
g-consumer      localhost:9092  (1)       uniform              Stable               2
--- kafka-consumer-groups.sh --describe --members

GROUP           CONSUMER-ID            HOST            CLIENT-ID       #PARTITIONS     CURRENT-EPOCH   CURRENT-ASSIGNMENT   TARGET-EPOCH    TARGET-ASSIGNMENT   
g-consumer      dHHCFy_6RmWBvpyk1U7leQ /127.0.0.1      g-consumer-A    3               3               orders:0,1,2         3               orders:0,1,2        
g-consumer      90SoJ7ajR6eTABfmOMF6Fw /127.0.0.1      g-consumer-B    3               3               orders:3,4,5         3               orders:3,4,5

• The classic group's assignment strategy is range (the client assignor's name); the consumer group's is uniform (the broker assignor's name).
• The member ids look different. In classic the broker builds them as client.id + "-" + UUID (ClassicGroup.generateMemberId(), lines 782–785); in the consumer protocol it's a 22-character UUID (base64) generated by the client.
• Only the consumer group shows CURRENT-EPOCH, TARGET-EPOCH and a target assignment. The tool's output, too, reveals that the broker keeps a separate "target" and "current" for each member. For the classic group those columns are -.

6. CoordinatorRuntime — write state to the log, rebuild it from the log

In section 5, GroupMetadataManager only returned a list of records and a response; it never wrote to disk. Writing is done by CoordinatorRuntime in the coordinator-common module, a shared runtime used by the group coordinator and the share-group coordinator (the transaction coordinator of post 12 is still Scala code that doesn't use it).

The runtime keeps one state machine per __consumer_offsets partition. When the broker becomes the leader of a partition, it creates that partition's state machine, and every request for the groups in it is handled there. Work goes into an event queue and is run by group-coordinator-event-processor-N threads. Their number is group.coordinator.threads (default 4, GroupCoordinatorConfig.java line 103), and only one event per partition runs at a time (the javadoc of EventAccumulator, lines 35–45). That is why there are almost no locks inside GroupMetadataManager — a group's state is only ever touched by one thread at a time.

Write operations run in CoordinatorWriteEvent.run().

        public void run() {
            try {
                // Get the context of the coordinator or fail if the coordinator is not in active state.
                withActiveContextOrThrow(tp, context -> {
                    // Execute the operation.
                    result = op.generateRecordsAndResult(context.coordinator.coordinator());

                    // Append the records and replay them to the state machine.
                    context.append(
                        producerId,
                        producerEpoch,
                        verificationGuard,
                        result.records(),
                        result.replayRecords(),
                        result.isAtomic(),
                        this
                    );

                    // ...
                });
            } catch (Throwable t) {
                complete(t);
            }
        }

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java lines 1317–1347

op.generateRecordsAndResult(…) is where a method like section 5's consumerGroupHeartbeat() gets called, and context.append(…) handles the result. The order inside append() is the core of this design.

                    try {
                        if (replay) {
                            coordinator.replay(
                                currentBatch.nextOffset,
                                producerId,
                                producerEpoch,
                                recordToReplay
                            );
                        }

                        currentBatch.builder.append(recordToAppend);
                        currentBatch.nextOffset++;

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java lines 1008–1019

1. The record is first applied to the in-memory state (replay()). For a consumer-protocol group, GroupMetadataManager doesn't modify state directly when it handles a request; it only produces records that say "this is how it should change", and the state changes in replay(). (Classic groups are the exception: member additions and removals and state transitions such as PreparingRebalance are applied directly to the ClassicGroup object — e.g. group.transitionTo(PREPARING_REBALANCE) in prepareRebalance(), line 7979 — and the result is recorded as one GroupMetadata record when the generation changes.)
2. The same record is appended to a batch (MemoryRecordsBuilder, the record batch of post 6).
3. When the batch is flushed (flushCurrentBatch()), partitionWriter.append(…) writes it to __consumer_offsets-N (lines 712–717). This takes the same road as a producer's records — through ReplicaManager to UnifiedLog.
4. The response isn't sent yet. The event goes into deferredEventQueue with the batch's offset (line 738), and completes once the partition's high watermark passes that offset (onHighWatermarkUpdated() → deferredEventQueue.completeUpTo(newHighWatermark), lines 1810–1831). The client hears "done" only when the records are replicated to the whole ISR and can't be lost even if the leader changes.

What if the in-memory state was changed first and then the write fails? The in-memory state is built from revertible data structures such as TimelineHashMap (the structures on top of SnapshotRegistry seen in post 10). Each time a batch is written to the log, a snapshot is taken at that offset (SnapshottableCoordinator.updateLastWrittenOffset(), lines 139–148), and if the next batch fails, failCurrentBatch() rolls back to that snapshot with revertLastWrittenOffset(batch base offset) (CoordinatorRuntime.java lines 786–792, SnapshottableCoordinator.java lines 85–96). Snapshots older than the high watermark are no longer needed and are deleted (deleteSnapshotsUpTo() in updateLastCommittedOffset(), lines 158–172). The controller does the same with the same SnapshotRegistry for the metadata log (metadata/…/controller/OffsetControlManager.java lines 271 and 328).

Offset commits take exactly the same road. GroupCoordinatorService.commitOffsets() calls scheduleWriteOperation() under the name "commit-offset" (lines 2374–2384), and OffsetMetadataManager.commitOffset() returns one GroupCoordinatorRecordHelpers.newOffsetCommitRecord(group, topic, partition, offset) record per committed partition (lines 665–670). When commitSync() returns, it means these records were written to __consumer_offsets and passed the high watermark.

Yellow boxes are where state lives. The dotted path is the recovery path when the broker restarts or becomes the partition's leader.

Rebuilding from the log

Because the log holds every change to the state, reading the log from the start and calling replay() gives the same state again (a classic group comes back from its last GroupMetadata record with that generation's members, as Empty or Stable — a rebalance in progress is not in the log, line 7094). When a broker becomes the leader of a __consumer_offsets partition (BrokerMetadataPublisher calls groupCoordinator.onElection, line 158), the runtime puts that partition's state machine into LOADING and lets CoordinatorLoaderImpl read it.

            long currentOffset = log.logStartOffset();
            LoadStats stats = new LoadStats();

            long lastCommittedOffset = -1L;
            while (shouldFetchNextBatch(currentOffset, logEndOffset(tp), stats.readAtLeastOneRecord)) {
                FetchDataInfo fetchDataInfo = log.read(currentOffset, loadBufferSize, FetchIsolation.LOG_END, true);

                stats.readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes() > 0;

                // ...
                MemoryRecords memoryRecords = toReadableMemoryRecords(tp, fetchDataInfo.records, buffer);
                // ...

                ReplayResult replayResult = processMemoryRecords(tp, log, memoryRecords, coordinator, stats, currentOffset, lastCommittedOffset);
                currentOffset = replayResult.nextOffset;
                lastCommittedOffset = replayResult.lastCommittedOffset;
            }

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java lines 138–156

It reads from the log start offset to the end with UnifiedLog.read() (the same method we saw in post 8) and calls coordinator.replay(…) for every record (inside processMemoryRecords(), lines 281–286). When it's done, the state machine becomes ACTIVE and logs "Finished loading of metadata from …" (CoordinatorRuntime.java lines 594–600). Until then, requests for groups in that partition get a COORDINATOR_LOAD_IN_PROGRESS error (CoordinatorRuntime.java lines 2096–2097), and clients retry a little later (the classic side's handling is at AbstractCoordinator.java lines 686–689).

The log doesn't grow forever because __consumer_offsets is a compacted topic (deleting older values of the same key and keeping only the last, post 7) — the topic config is at line 2610 of GroupCoordinatorService.java. Record keys are things like "group + member" or "group + topic + partition", so after compaction only the latest state per key remains. When a member leaves, a record with no value (a tombstone) is written for that member's keys so compaction removes them.

Example 3: the records inside __consumer_offsets

After example 1 finished, we decoded the partitions the two groups wrote to (18 and 32) with kafka-dump-log.sh --offsets-decoder. The tool prints keys and values as JSON, but the lines are long, so we replaced the key's type number with the record name and kept only a few fields, one line per record (OffsetCommit records are only counted; the fmt.awk used for this is in the example folder). The member ids, and the record numbers and OffsetCommit count (which depend on where auto-commits land), differ from run to run.

=== 4. __consumer_offsets records
--- g-classic -> __consumer_offsets-18
  0  GroupMetadata                                       generation=0 protocol=null leader=null members=0
  1  GroupMetadata                                       generation=1 protocol=range leader=g-classic-A-3.. members=1
  8  GroupMetadata                                       generation=2 protocol=range leader=g-classic-A-3.. members=2
 36  GroupMetadata                                       generation=3 protocol=range leader=g-classic-A-3.. members=1
 37  GroupMetadata                                       generation=4 protocol=null leader=null members=0
(+ 33 OffsetCommit records)
--- g-consumer -> __consumer_offsets-32
  0  ConsumerGroupMemberMetadata            member=dHHC  clientId=g-consumer-A
  1  ConsumerGroupMetadata                               groupEpoch=2
  2  ConsumerGroupTargetAssignmentMember    member=dHHC  target=[0,1,2,3,4,5]
  3  ConsumerGroupTargetAssignmentMetadata               assignmentEpoch=2
  4  ConsumerGroupCurrentMemberAssignment   member=dHHC  memberEpoch=2 state=0 assigned=[0,1,2,3,4,5] pending=
  5  ConsumerGroupMemberMetadata            member=90So  clientId=g-consumer-B
  6  ConsumerGroupMetadata                               groupEpoch=3
  7  ConsumerGroupCurrentMemberAssignment   member=90So  memberEpoch=2 state=0 assigned= pending=
  8  ConsumerGroupTargetAssignmentMember    member=dHHC  target=[0,1,2]
  9  ConsumerGroupTargetAssignmentMember    member=90So  target=[3,4,5]
 10  ConsumerGroupTargetAssignmentMetadata               assignmentEpoch=3
 11  ConsumerGroupCurrentMemberAssignment   member=dHHC  memberEpoch=2 state=1 assigned=[0,1,2] pending=[3,4,5]
 24  ConsumerGroupCurrentMemberAssignment   member=90So  memberEpoch=3 state=2 assigned= pending=
 25  ConsumerGroupCurrentMemberAssignment   member=dHHC  memberEpoch=3 state=0 assigned=[0,1,2] pending=
 26  ConsumerGroupCurrentMemberAssignment   member=90So  memberEpoch=3 state=0 assigned=[3,4,5] pending=
 51  ConsumerGroupCurrentMemberAssignment   member=90So  <DELETE>
 52  ConsumerGroupTargetAssignmentMember    member=90So  <DELETE>
 53  ConsumerGroupMemberMetadata            member=90So  <DELETE>
 54  ConsumerGroupMetadata                               groupEpoch=4
 55  ConsumerGroupTargetAssignmentMember    member=dHHC  target=[0,1,2,3,4,5]
 56  ConsumerGroupTargetAssignmentMetadata               assignmentEpoch=4
 57  ConsumerGroupCurrentMemberAssignment   member=dHHC  memberEpoch=4 state=0 assigned=[0,1,2,3,4,5] pending=
 64  ConsumerGroupCurrentMemberAssignment   member=dHHC  <DELETE>
 65  ConsumerGroupTargetAssignmentMember    member=dHHC  <DELETE>
 66  ConsumerGroupMemberMetadata            member=dHHC  <DELETE>
 67  ConsumerGroupMetadata                               groupEpoch=5
 68  ConsumerGroupTargetAssignmentMetadata               assignmentEpoch=5
(+ 42 OffsetCommit records)

The key's type number is the apiKey of the record schema JSON, which I checked in group-coordinator/src/main/resources/common/message/ — OffsetCommitKey (1), GroupMetadataKey (2), ConsumerGroupMetadataKey (3), ConsumerGroupMemberMetadataKey (5), ConsumerGroupTargetAssignmentMetadataKey (6), ConsumerGroupTargetAssignmentMemberKey (7), ConsumerGroupCurrentMemberAssignmentKey (8). Like the requests and responses of post 3, these schemas are turned into Java classes at build time.

• A classic group is one GroupMetadata record per group. Every time the generation changes, a value holding all members and their assignments is written again in full. The assignment computed by the leader (g-classic-A…) is stored as bytes, untouched.
• A consumer group writes its state in small pieces. The group epoch (ConsumerGroupMetadata), member info (MemberMetadata), each member's target (TargetAssignmentMember) and the target epoch (TargetAssignmentMetadata), and each member's current assignment (CurrentMemberAssignment) all have different keys. Even the target assignment is written only for members whose target changed (the comment at lines 2705–2706). It doesn't rewrite the whole group in one value per generation, as classic does.
• The state numbers are the values of modern/MemberState.java — 0 = STABLE, 1 = UNREVOKED_PARTITIONS, 2 = UNRELEASED_PARTITIONS (lines 32–45). The state diagram of section 5, left behind as records.
• Records 0–4 are A joining, 5–7 are B joining. B's arrival raised the group epoch to 3 (record 6), but the target assignment wasn't recomputed and B's current assignment is the empty assignment of epoch 2 (record 7). In the raw dump lines, the time A's target was computed (assignmentTimestamp of record 3) and the time B's records were written (CreateTime) were 344 ms apart. group.consumer.assignment.interval.ms (1000 ms) hadn't passed, so the computation was postponed (section 5).
• Records 8–11 were written about 4.9 seconds later, on A's regular heartbeat. The new target (A=[0,1,2], B=[3,4,5], epoch 3) is computed, and since A has to give up 3, 4 and 5 it stays at epoch 2 with state=1 (UNREVOKED).
• Record 24: B moves to epoch 3 but A still holds 3, 4 and 5, so it's state=2 (UNRELEASED) with an empty assignment. Record 25: 7 ms later A reports letting go and becomes STABLE. Record 26: 5 seconds later, on B's next heartbeat, B gets 3, 4 and 5. This is why B took 10 seconds to get [3, 4, 5] in example 1.
• Records 51–53: as B leaves (a heartbeat with epoch -1), <DELETE> (tombstones) are written for B's three keys, the group epoch goes up to 4, and A gets all six. Records 64–68 are A leaving. Even with no members, the group and its committed offsets remain (the output of example 4).
• The gaps in the numbering (12–23 and so on) are OffsetCommit records written by auto-commit.

Example 4: restarting the broker

Finally, we shut the server down cleanly, started it again, and looked at the load log lines of the two partitions and the committed offsets of g-consumer. At the end we pulled only the group coordinator thread names out of the broker's jstack.

=== 5. restart the server: the coordinator replays its partitions
Finished loading of metadata from __consumer_offsets-18 with epoch 2 in 290ms where 248ms was spent in the scheduler. Loaded 38 records which total to 4268 bytes. (org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime)
Finished loading of metadata from __consumer_offsets-32 with epoch 2 in 246ms where 7ms was spent in the scheduler. Loaded 69 records which total to 6233 bytes. (org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime)

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
g-consumer      orders          0          10              10              0               -               -               -
g-consumer      orders          1          9               9               0               -               -               -
g-consumer      orders          2          13              13              0               -               -               -
g-consumer      orders          3          6               6               0               -               -               -
g-consumer      orders          4          14              14              0               -               -               -
g-consumer      orders          5          8               8               0               -               -               -
--- group coordinator threads in the broker (jstack)
"group-coordinator-event-processor-0"
"group-coordinator-event-processor-1"
"group-coordinator-event-processor-2"
"group-coordinator-event-processor-3"
"group-coordinator-reaper"
stopped

Loaded N records is the number of records read from the start of the partition and fed to replay(). It equals the count from example 3 (group records + OffsetCommit); the count, time and bytes differ from run to run. The group state in memory vanished with the server, but it was rebuilt from the log as it was, and the committed offsets are the same as before the restart. The threads are group-coordinator-event-processor-0 to -3 (4 by default, GroupCoordinatorService.java lines 263–269) and group-coordinator-reaper, which drives timers such as session expiry and rebalance timeouts (BrokerServer.scala lines 695–698). The group-coordinator-background-… pool for heavy work such as resolving regular expressions (lines 271–276) doesn't appear: that pool extends the JDK's ThreadPoolExecutor, which creates a thread when the first task arrives, and this example never gave it one.

7. Settings at a glance

Here are the settings that came up. Every default was checked against the constant passed to define(…).

ConsumerConfig.java lines 460–499, GroupCoordinatorConfig.java lines 460–494

The settings from the table above, regrouped by where they are used. All values are the defaults from the table.

For the broker to accept the consumer protocol, the feature flag group.version must be 1 or higher and the broker setting group.coordinator.rebalance.protocols must include consumer (KafkaApis.isConsumerGroupProtocolEnabled(), lines 2675–2678). That setting's default contains classic, consumer and streams, and it is marked deprecated for removal in 5.0 (GroupCoordinatorConfig.java lines 82–92). The example's server was freshly created with kafka-storage.sh format, and the consumer-protocol run succeeded, so it was on from the start (the example's kafka-features.sh describe output is cut to its first six fields by run.sh, so it shows only the supported range up to SupportedMaxVersion: 1, not the finalized level). A cluster upgraded from an older version may need it raised separately (what I verified in the source is only this condition).

What this post leaves out

• Classic and new members mixed in one group (online migration, group.consumer.migration.policy): GroupMetadataManager.classicGroupJoinToConsumerGroup() from line 2763, ConsumerGroupMigrationPolicy.java
• Static membership (group.instance.id): GroupMetadataManager.getOrMaybeSubscribeStaticConsumerGroupMember(), staticMembers in ClassicGroup.java
• Regex subscriptions (subscribe(Pattern), SubscriptionPattern) and regex resolution on the broker: modern/consumer/TopicRegexResolver.java, GroupMetadataManager.maybeUpdateRegularExpressions()
• The assignor algorithms themselves: group-coordinator/…/assignor/UniformAssignor.java (and uniform2/), RangeAssignor.java, and on the client CooperativeStickyAssignor.java·AbstractStickyAssignor.java
• Offset expiry and group deletion: OffsetExpirationConditionImpl.java, OffsetMetadataManager.cleanupExpiredOffsets()
• The consumer's position tracking and interpreting Fetch results: ConsumerSubscriptionState.java, FetchCollector.java, OffsetFetcher.java·OffsetsRequestManager.java (continues from post 8)
• Transactional offset commits (TxnOffsetCommit, OffsetMetadataManager.commitTransactionalOffset()): post 12
• Kafka Streams groups (streams/, StreamsGroupHeartbeat) and share groups (share-coordinator/, KafkaShareConsumer, ShareGroupHeartbeat) — left out of the whole series

Summary

What's next

Committing an offset and processing a record are two separate actions. Crash after processing but before committing and you read the same record again; crash after committing but before processing and you lose it. A producer, too, can write the same record twice while retrying. The last post looks at the mechanisms that close these gaps — the idempotent producer that blocks duplicates with a producer id and sequence numbers, transactions that bundle writes to several partitions together with an offset commit (sendOffsetsToTransaction() is the road into this post's __consumer_offsets), and the consumer's isolation.level=read_committed — in post 12: Idempotence and Transactions — How Exactly-Once Is Built, and Wrapping Up the Series.

The idea in this post — "write state to the log as records, then read them back to rebuild it" — is the same idea as what the controller of post 10 does with the metadata log. Read the two side by side and you'll see that Kafka trusts its own state to Kafka's log, too.

• Post 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Post 4: The Broker's Network Layer — Acceptors, Processors and Request Handlers
• Post 8: The Read Path — Fetch Requests, Long Polling and Zero-Copy
• Post 9: Replication — The ISR, the High Watermark and the Wait Behind acks=all
• Post 10: KRaft and Metadata — How the Controller Remembers the Cluster