Digging into Kafka's Source 10: KRaft and Metadata — How the Controller Remembers the Cluster

What this post covers

This is the tenth post in "Digging into Kafka's Source" and the first of the third group, "what keeps the cluster running". From Part 3 to Part 9 we followed the data path: one record going from the producer to the broker's disk and on to followers and consumers. Now we take the ledger path that Part 2 only named. Who decides facts like "topic orders has 3 partitions, the leader of partition 0 is broker 1, its ISR is 1, 2, 3", where are they written down, and how do the brokers find out?

This post zooms into region G of the Part 2 map, "Controller and metadata (KRaft)". Blue boxes are the parts we read here, yellow is where the ledger is stored, grey boxes were covered in earlier posts.

Part 2 showed the big flow of creating one topic: the request is forwarded from a broker to the controller, the controller appends a TopicRecord and PartitionRecords to the metadata log, and the brokers read that log and create the folders. This post opens each box in that flow. It tries to answer four questions.

• What is the cluster's ledger (__cluster_metadata) made of, and when there are several controllers, who writes it and how do the others end up with the same content?
• When a controller dies, how is a new leader elected, and when exactly is something "committed"?
• How does the controller handle requests without locks, and how does it undo decisions that have not committed yet?
• How do brokers receive and apply changes to the ledger, and how does the controller notice a dead broker?

All line numbers in this post refer to this commit. How to get the same source and build the Docker image used by the examples is in Part 1. The code read here is almost identical to 4.4.0, the latest stable release most people use. Comparing the classes above with the 4.4.0 tag, QuorumState, KafkaRaftLog, MetadataLoader, MetadataDelta, MetadataImage, KafkaEventQueue, OffsetControlManager, SnapshotRegistry, TimelineHashMap and BrokerLifecycleManager do not differ by a single line; KafkaRaftClient and LeaderState differ only in the code that changes the voter set (add/remove/update voter), and QuorumController and ReplicationControlManager only in taking the cap on records one request may generate from a new setting, controller.max.records.per.batch (KRaftConfigs.java line 72), plus, in QuorumController, a small change in how it checks node ids.

The code in this post is almost all Java. KRaft was written as Java modules from the start (raft, metadata, server-common); the Scala core module only keeps the shells that wire them together. The files are big, though (KafkaRaftClient.java is 4,185 lines, QuorumController.java 2,245). This post picks the main line a first-time reader can follow; what I left out is listed by name and location under "What this post leaves out" at the end.

Terms to know first

Metadata: in this post, not the data (records) but every fact about the cluster. Which brokers are alive, which topics exist and how many partitions they have, the replicas, leader and ISR of every partition, topic configs, quotas, ACLs. This is what Part 2 called "the ledger".

Metadata log and metadata records: the ledger is not a table but a log of changes in time order. Records such as "a topic was created" (TopicRecord), "a partition was created like this" (PartitionRecord), "a partition's ISR changed" (PartitionChangeRecord) and "broker 3 was fenced" (BrokerRegistrationChangeRecord) are appended one after another. Replaying them from the start gives you the current ledger, just as adding up a bank statement from the first line gives you the balance. This log is partition 0 of the topic __cluster_metadata.

Quorum, voters, observers: the group of controllers that keep the metadata log together is the quorum, and its members are voters. Only voters vote in leader elections and count toward the majority that decides whether a record is "committed". Brokers are observers: they follow the log without a vote (the CurrentObservers in the Part 2 example).

Epoch: a generation number that goes up by one every time the leader changes — the same idea as a data partition's leader epoch (Part 9). Requests carry the epoch, so a late request from an old leader is recognised as "last generation" and dropped.

Commit and high watermark (HW): a record the leader has written is committed once it is replicated to a majority of voters. Since a majority has it, even if the leader dies the new leader (which also comes from a majority) is guaranteed to have it. The offset that separates committed from uncommitted records is the high watermark. For data partitions, as we saw in Part 9, the high watermark is based on the whole ISR; for the metadata log it is based on a majority. Section 2 shows the difference in the source.

Active controller and standby controllers: the Raft leader of the metadata log is the active controller. Only the active controller makes decisions (new records); standby controllers receive those records, apply them to their own memory, and take over right away when leadership changes. We saw this in the QuorumController javadoc in section 4 of Part 2.

Fence: a mark the controller puts on a broker meaning "don't trust this broker right now". A fenced broker is not given leadership and is removed from ISRs. It is the fence Part 2 mentioned: a starting broker waits "until the fence is lifted". Section 6 shows when the controller puts the fence up and takes it down.

1. The ledger is an ordinary Kafka log — __cluster_metadata-0

The metadata log sits on UnifiedLog too

The metadata log's name is one constant in clients (CLUSTER_METADATA_TOPIC_NAME = "__cluster_metadata", Topic.java line 30). The log is held by the KafkaRaftManager that SharedServer creates (section 3 of Part 2).

  override val raftLog: RaftLog = buildMetadataLog()
  private val netChannel = buildNetworkChannel()
  private val expirationTimer = new SystemTimer("raft-expiration-executor")
  private val expirationService = new TimingWheelExpirationService(expirationTimer)
  override val client: KafkaRaftClient[T] = buildRaftClient()
  private val clientDriver = new KafkaRaftClientDriver[T](client, threadNamePrefix, fatalFaultHandler, logContext)

core/src/main/scala/kafka/raft/KafkaRaftManager.scala lines 129–134

Reading the Scala — val is a field that cannot be reassigned (a Java final field). override val raftLog: RaftLog = … fills the raftLog member of the RaftManager interface this class implements, and the [T] in KafkaRaftClient[T] is Java's <T>.

These three parts are sections 1–2 of this post: raftLog is the log on disk, client (KafkaRaftClient) is the Raft protocol, and clientDriver is the thread that runs the protocol. Open the KafkaRaftLog that buildMetadataLog() creates, and a familiar class shows up.

    @Override
    public LogAppendInfo appendAsLeader(Records records, int epoch) {
        if (records.sizeInBytes() == 0) {
            throw new IllegalArgumentException("Attempt to append an empty record set");
        }

        try {
            return handleAndConvertLogAppendInfo(log.appendAsLeader((MemoryRecords) records, epoch, AppendOrigin.RAFT_LEADER));
        } catch (IOException ioe) {
            throw new UncheckedIOException(ioe);
        }
    }

raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftLog.java lines 147–158

The log here is private final UnifiedLog log; (line 84) — the very UnifiedLog that data partitions use, which we saw in Part 6 and Part 7. So the metadata log also has segment files (.log), an offset index (.index) and a time index (.timeindex), and its records are stored in the same record batch v2 format as data. That is the raft → storage arrow in the module diagram of Part 2. The ledger is also a Kafka log — KRaft's biggest design decision fits in that one line. It means storage, replication and tools (kafka-dump-log.sh) are shared with data.

Metadata records are generated from JSON schemas too

Like the requests and responses in Part 3, records are defined by JSON schemas and Java classes are generated from them at build time. There are 27 schemas in metadata/src/main/resources/common/metadata/ (TopicRecord.json, PartitionRecord.json, PartitionChangeRecord.json, RegisterBrokerRecord.json, BrokerRegistrationChangeRecord.json, FenceBrokerRecord.json, ConfigRecord.json, FeatureLevelRecord.json, NoOpRecord.json … and, by name only, ZkMigrationRecord.json from the days of migrating off ZooKeeper). Here is the shortest, TopicRecord.

{
  "apiKey": 2,
  "type": "metadata",
  "name": "TopicRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
      "about": "The topic name." },
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The unique ID of this topic." }
  ]
}

metadata/src/main/resources/common/metadata/TopicRecord.json lines 16–28

Where a request schema says "type": "request", this says "type": "metadata", and apiKey is the record type number. From this file the generator makes the TopicRecord class and the MetadataRecordType enum that maps numbers to types (both are generated at build time, so neither is in the repository). PartitionRecord.json has fields such as the partition id, topic id, Replicas, Isr, Leader, LeaderEpoch and PartitionEpoch (PartitionRecord.json lines 24–53). The values buildPartitionRegistration() filled in, in section 5 of Part 2, become exactly these fields.

Example 1: create a topic and read the metadata log directly

This example was run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411); the same goes for every example in this post. We start the broker and controller in one process (node 1), check the quorum, create the topic orders (3 partitions), and look at the metadata log folder and the records in it. The script is kafka-examples/ex/u10-metadata-log/run.sh; the tools are kafka-metadata-quorum.sh … describe --status, ls and cat, and kafka-dump-log.sh --cluster-metadata-decoder --files …/00000000000000000000.log, which decodes metadata records. The first part of the output (cluster id, directory ids, topic ids, offsets, timestamps and the like differ on every run):

=== quorum status (kafka-metadata-quorum.sh describe --status)
ClusterId:              cTyAk73kQ4CIt8MyqMzHOQ
LeaderId:               1
LeaderEpoch:            1
HighWatermark:          32
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   0
CurrentVoters:          [{"id": 1, "directoryId": "5H7XVkyAS9mZZQhe8CMRaw", "endpoints": ["CONTROLLER://localhost:9093"]}]
CurrentObservers:       []
=== create topic orders (3 partitions)
Created topic orders.
=== files in __cluster_metadata-0
00000000000000000000-0000000000.checkpoint
00000000000000000000.index
00000000000000000000.log
00000000000000000000.timeindex
leader-epoch-checkpoint
partition.metadata
quorum-state
=== quorum-state file
{"leaderId":1,"leaderEpoch":1,"votedId":-1,"votedDirectoryId":"AAAAAAAAAAAAAAAAAAAAAA","data_version":1}

The folder looks just like the data partition folder (orders-0) from Part 7. Two files differ. quorum-state is where Raft records "the current epoch and leader, and whom I voted for in this epoch" (FileQuorumStateStore.DEFAULT_FILE_NAME = "quorum-state", FileQuorumStateStore.java line 80); we meet it again in the election in section 2. …-0000000000.checkpoint is a snapshot file. Its name is <end offset>-<epoch>, and this one is the "bootstrap" snapshot (offset 0, epoch 0) written by kafka-storage.sh format. Snapshots come in section 5.

Inside the log, the records by type looked like this:

=== metadata log: the first batch (control records written by the new leader)
baseOffset: 0 lastOffset: 2 count: 3 ... isControl: true
 offset: 0 LeaderChange: {"version":0,"leaderId":1,"voters":[{"voterId":1}],"grantingVoters":[{"voterId":1}]}
 offset: 1 KRaftVersion {"version":0,"kRaftVersion":1}
 offset: 2 KRaftVoters {"version":0,"voters":[{"voterId":1,"voterDirectoryId":"5H7XVkyAS9mZZQhe8CMRaw","endpoints":[{"name":"CONTROLLER","host":"localhost","po
=== metadata log: record types (count)
     32 "type":"NO_OP_RECORD"
      6 "type":"FEATURE_LEVEL_RECORD"
      3 "type":"PARTITION_RECORD"
      2 "type":"BROKER_REGISTRATION_CHANGE_RECORD"
      1 "type":"TOPIC_RECORD"
      1 "type":"REGISTER_CONTROLLER_RECORD"
      1 "type":"REGISTER_BROKER_RECORD"
      1 "type":"END_TRANSACTION_RECORD"
      1 "type":"CONFIG_RECORD"
      1 "type":"BEGIN_TRANSACTION_RECORD"
=== metadata log: records for topic orders (the batch header, then the records)
baseOffset: 41 lastOffset: 44 count: 4 ... isControl: false
 offset: 41 {"type":"TOPIC_RECORD","version":0,"data":{"name":"orders","topicId":"oerNM_RtQJiR5zDk9XwK7w"}}
 offset: 42 {"type":"PARTITION_RECORD","version":2,"data":{"partitionId":0,"topicId":"oerNM_RtQJiR5zDk9XwK7w","replicas":[1],"isr":[1], ...,"directories":["5H7XVkyAS9mZZ
 offset: 43 {"type":"PARTITION_RECORD","version":2,"data":{"partitionId":1,"topicId":"oerNM_RtQJiR5zDk9XwK7w","replicas":[1],"isr":[1], ...,"directories":["5H7XVkyAS9mZZ
 offset: 44 {"type":"PARTITION_RECORD","version":2,"data":{"partitionId":2,"topicId":"oerNM_RtQJiR5zDk9XwK7w","replicas":[1],"isr":[1], ...,"directories":["5H7XVkyAS9mZZ
=== controller.log: the controller replayed the records
04:55:26,907 INFO [QuorumController id=1] Replayed TopicRecord for topic orders with topic ID oerNM_RtQJiR5zDk9XwK7w. (org.apache.kafka.cont
04:55:26,912 INFO [QuorumController id=1] Replayed PartitionRecord for new partition orders-0 with topic ID oerNM_RtQJiR5zDk9XwK7w and Parti

(The script truncates long lines. The ... stands for the removingReplicas, addingReplicas, leader, leaderEpoch and partitionEpoch fields. The last two lines are the Replayed … lines picked from controller.log, left by the replay() of section 3. The describe --replication output the script prints next (one line, the leader) is left out.)

Three things to read here:

1. One topic = one record batch. One TopicRecord and three PartitionRecords sit in a single batch at offsets 41–44. The record list built by createTopics() in section 5 of Part 2 was appended in one go.
2. Control records come first. Offsets 0–2 are an isControl: true batch: the LeaderChange and voter list this node wrote when it became leader of epoch 1. Section 2 explains why they are at the very front.
3. NO_OP_RECORD is the most common record. Even when nothing happens, the active controller writes an empty record every metadata.max.idle.interval.ms (default 500, MetadataLogConfig.java lines 75–76; registerWriteNoOpRecord(...), QuorumController.java lines 1634–1636). The log has to keep moving so brokers can keep reporting "I have caught up to offset N" and the high watermark keeps advancing. The longer the run, the more of these lines.

The ledger is an ordinary log on top of UnifiedLog. Offsets, record types and counts come from Example 1's output above (offsets vary per run).

2. KafkaRaftClient — elections are Raft, replication is Fetch

What the class description sums up

The javadoc of KafkaRaftClient starts by saying where this implementation matches textbook Raft and where it does not.

/**
 * This class implements a Kafkaesque version of the Raft protocol. Leader election
 * is more or less pure Raft, but replication is driven by replica fetching and we use Kafka's
 * log reconciliation protocol to truncate the log to a common point following each leader
 * election.
 *
 * Like Zookeeper, this protocol distinguishes between voters and observers. Voters are
 * the only ones who are eligible to handle protocol requests and they are the only ones
 * who take part in elections.
 *
 * These are the APIs in this protocol:
 * ...
 * 2) {@link BeginQuorumEpochRequestData}: Sent by the leader of an epoch only to valid voters to
 *    assert its leadership of the new epoch. This request will be retried indefinitely for
 *    each voter until it acknowledges the request or a new election occurs.
 *
 *    This is not needed in usual Raft because the leader can use an empty data push
 *    to achieve the same purpose. The Kafka Raft implementation, however, is driven by
 *    fetch requests from followers, so there must be a way to find the new leader after
 *    an election has completed.
 * ...
 */

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java lines 129–168

(The omitted parts describe the other APIs — 1) Vote, 3) EndQuorumEpoch, 4) Fetch and 5) FetchSnapshot.)

In short: in textbook Raft the leader pushes records to followers (AppendEntries). Kafka does the opposite — followers pull from the leader (Fetch), just like follower replication in Part 9 and consumers in Part 8. Since the leader then has no way to speak first, there is a separate BeginQuorumEpoch to announce "I am the new leader" after an election. The javadoc lists five requests — Vote, BeginQuorumEpoch, EndQuorumEpoch, Fetch, FetchSnapshot (DescribeQuorum, which asks for the quorum state, and the voter-change requests come in the same way) — and they sit side by side in the match of the controller's dispatcher ControllerApis.handle(), like case ApiKeys.VOTE => handleVote(request) (ControllerApis.scala lines 96–109; match was explained in Part 2). All of these branches funnel into handleRaftRequest(), which calls raftManager.handleRequest(...) (lines 707–714).

One thread runs poll()

KafkaRaftClientDriver appends -io-thread to the thread name (line 55), creating a thread like kafka-1-raft-io-thread, and that thread does only this:

    @Override
    public void doWork() {
        try {
            client.poll();
        } catch (Throwable t) {
            throw fatalFaultHandler.handleFault("Unexpected error in raft IO thread", t);
        }
    }

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClientDriver.java lines 61–68

The parent class ShutdownableThread calls doWork() over and over until told to stop. One poll() means "do whatever the current role requires, handle one incoming message, notify the listeners".

    public void poll() {
        // ...
        long pollStateTimeoutMs = pollCurrentState(startPollTimeMs);
        long cleaningTimeoutMs = snapshotCleaner.maybeClean(startPollTimeMs);
        long pollTimeoutMs = Math.min(pollStateTimeoutMs, cleaningTimeoutMs);

        // ...
        var maybeEntry = messageQueue.poll(pollTimeoutMs);

        long endWaitTimeMs = time.milliseconds();
        // ...
        maybeEntry.ifPresent(entry -> handleInboundMessage(entry.message(), entry.future(), endWaitTimeMs));

        pollListeners();
    }

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java lines 3687–3712

1. pollCurrentState() — depending on the role, calls one of pollLeader(), pollCandidate(), pollProspective(), pollFollower(), pollUnattached(), pollResigned() (the if … else if chain at lines 3540–3556). A leader writes accumulated records to the log and sends BeginQuorumEpoch; a follower sends Fetch to the leader; a candidate sends Vote. It returns "how long until there is something to do next".
2. messageQueue.poll(pollTimeoutMs) — waits that long for incoming requests and responses. Raft requests received by the controller's request handler threads and responses to requests this node sent both arrive in this queue.
3. pollListeners() — if there are newly committed batches, notifies the listeners (the QuorumController on active and standby controllers, the MetadataLoader on brokers) through handleCommit(). Sections 3 and 5 start here.

All Raft state is changed by this one thread only. The thread that receives a request just puts the message in the queue and takes back a future (messageQueue.add(...), KafkaRaftClient.java lines 3672–3681). Unlike the request handler pool in Part 4, Raft lives and dies by ordering, so everything is put in a single line.

Roles and state transitions

The QuorumState javadoc lists how the role can change, almost like a table (QuorumState.java lines 35–83). As a picture:

Role states of a voter (controller). To keep the picture simple I left out a few arrows that leave almost every state, such as "learns of an election with a higher epoch → Unattached". Observers (brokers) only move between Unattached and Follower.

Each state has its own class (UnattachedState, ProspectiveState, CandidateState, LeaderState, FollowerState, ResignedState). What textbook Raft does not have is Prospective. A follower that hears nothing from the leader for controller.quorum.fetch.timeout.ms (default 2,000ms, QuorumConfig.java lines 79–83) does not become a candidate right away; it first runs a PreVote ("if I held an election now, would you vote for me?"). This practice vote does not raise the epoch, so a node that was briefly cut off cannot come back, bump the epoch for nothing and drag down a perfectly healthy leader. Only when a majority says yes does it become a Candidate, raise the epoch by one (int newEpoch = epoch() + 1;, QuorumState.java line 641) and send real Votes.

State changes are written to disk first.

    private void durableTransitionTo(EpochState newState) {
        log.info("Attempting durable transition to {} from {}", newState, state);
        store.writeElectionState(newState.election(), partitionState.lastKraftVersion());
        memoryTransitionTo(newState);
    }

raft/src/main/java/org/apache/kafka/raft/QuorumState.java lines 729–733

store.writeElectionState() writes the quorum-state file we saw in Example 1. "I voted for 102 in epoch 2" has to be on disk first; otherwise a process that dies right after voting and comes back could vote for a different candidate in the same epoch, and one epoch could end up with two leaders. The Completed transition to … from … log line that memoryTransitionTo() writes (line 742) becomes our window onto the election in Example 2.

The one line that decides a vote

How a voter judges a Vote request lives in handleVoteRequest().

        if (replicaEpoch > quorum.epoch()) {
            transitionToUnattached(replicaEpoch, OptionalInt.empty());
        }
        // ...
        OffsetAndEpoch lastEpochEndOffsetAndEpoch = new OffsetAndEpoch(lastEpochEndOffset, lastEpoch);
        ReplicaKey replicaKey = ReplicaKey.of(
            replicaId,
            partitionRequest.replicaDirectoryId()
        );
        boolean voteGranted = quorum.canGrantVote(
            replicaKey,
            lastEpochEndOffsetAndEpoch.compareTo(endOffset()) >= 0,
            preVote
        );

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java lines 896–929

(The omitted part checks that the request was meant for this node.)

1. If the request's epoch is larger than mine, I first move to Unattached in that epoch (a new election has started).
2. lastEpochEndOffsetAndEpoch.compareTo(endOffset()) >= 0 — a vote can be granted only if the candidate's log is not behind mine, compared as (last epoch, end offset). Committed records are held by a majority, and a candidate needs a majority of votes, so this single rule guarantees that "a node that doesn't know a committed record can never become leader".
3. When a real (non-PreVote) vote is granted, the code right below records it (unattachedAddVotedState(...) inside if (!preVote && voteGranted), lines 931–937 → the durableTransitionTo above). PreVotes are not recorded.

The election winner writes control records right away in onBecomeLeader() (state.appendStartOfEpochControlRecords(currentTimeMs); the comment says "the high watermark can only be advanced once we have written a record from the new leader's epoch, hence we write a control message immediately", KafkaRaftClient.java lines 670–673). That is the LeaderChange at offset 0 in Example 1. Why a record "from its own epoch" is needed before the high watermark can move is explained just below.

The fetch offset doubles as "I got it"

The heart of tryCompleteFetchRequest(), where the leader handles a follower's Fetch:

            final Records records;
            if (validOffsetAndEpoch.kind() == ValidOffsetAndEpoch.Kind.VALID) {
                LogFetchInfo info = log.read(fetchOffset, Isolation.UNCOMMITTED, maxSizeBytes);

                if (state.updateReplicaState(replicaKey, currentTimeMs, info.startOffsetMetadata)) {
                    onUpdateLeaderHighWatermark(state, currentTimeMs);
                }

                records = info.records;
            } else {
                records = MemoryRecords.EMPTY;
            }

            return buildFetchResponse(
                // ...
                state.highWatermark()
            );

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java lines 1633–1653

When a follower says "give me records from offset N", it also means "I have everything before N". The leader records that follower's end offset as N with updateReplicaState(), and if the high watermark moved as a result, it announces it. The response carries the current high watermark along with the records, so the follower also learns how far things are committed. This is the same design as follower fetching for data partitions in Part 9. When there is no new data, the response is parked (fetchPurgatory.await(...), line 1551) and sent when data arrives or time runs out — the same long-polling idea as DelayedFetch in Part 8.

The high watermark is computed in LeaderState.maybeUpdateHighWatermark().

    private boolean maybeUpdateHighWatermark() {
        // Find the largest offset which is replicated to a majority of replicas (the leader counts)
        ArrayList<ReplicaState> followersByDescendingFetchOffset = followersByDescendingFetchOffset()
            .collect(Collectors.toCollection(ArrayList::new));

        int indexOfHw = voterStates.size() / 2;
        Optional<LogOffsetMetadata> highWatermarkUpdateOpt = followersByDescendingFetchOffset.get(indexOfHw).endOffset;

        if (highWatermarkUpdateOpt.isPresent()) {

            // The KRaft protocol requires an extra condition on commitment after a leader
            // election. The leader must commit one record from its own epoch before it is
            // allowed to expose records from any previous epoch. This guarantees that its
            // log will contain the largest record (in terms of epoch/offset) in any log
            // which ensures that any future leader will have replicated this record as well
            // as all records from previous epochs that the current leader has committed.

            LogOffsetMetadata highWatermarkUpdateMetadata = highWatermarkUpdateOpt.get();
            long highWatermarkUpdateOffset = highWatermarkUpdateMetadata.offset();

            if (highWatermarkUpdateOffset > epochStartOffset) {
                // ...

raft/src/main/java/org/apache/kafka/raft/LeaderState.java lines 660–680

It lines up the voters (the leader included) by end offset, largest first (ReplicaState.compareTo() sorts descending, lines 1025–1034), and picks the entry at size / 2. With 3 voters that is index 1 (counting from 0) — the offset of the second-furthest node, i.e. the largest offset that two out of three have. With 5 voters it is index 2, the offset three out of five have. It does not wait for the slowest one (two, with five voters).

Then highWatermarkUpdateOffset > epochStartOffset — a new leader does not move the high watermark until a record written in its own epoch has committed. Declaring older-epoch records committed just because they "seem to be on a majority" can, in a known case from the Raft paper, let a node without those records win the next election (that is the condition the comment describes). That is why onBecomeLeader() writes a control record immediately after winning — so this condition is met quickly.

The metadata log's high watermark is "the offset a majority has", and a new leader doesn't raise it until a record from its own epoch is committed.

Easy to confuse: the HW of a data partition and of the metadata log use different rules

Both use the same Fetch, but "who picks the leader" differs. A data partition gets its leader from outside (the controller), so it can use a flexible set called the ISR. The metadata log is the very bottom of the cluster with nothing outside to lean on, so it decides for itself by majority vote. That is why people usually run 3 or 5 controllers: with 3, one can die; with 5, two can, and a majority remains.

Example 2: kill the leader of three controllers and watch the election

We start three controllers (nodes 101, 102, 103 on ports 9093, 9095, 9097) and one broker (node 1). The series' shared helper (kafka_cluster3) has only one controller, so run.sh writes the config files itself and formats all three controllers with the same --initial-controllers "101@localhost:9093:<directory id>,102@…,103@…". After checking the quorum and creating a topic, it kills the leader controller with kill -9 and waits 15 seconds. The script is ex/u10-quorum3/run.sh.

=== quorum status
LeaderId:               101
LeaderEpoch:            1
HighWatermark:          57
CurrentVoters:          [{"id": 101, "directoryId": "_pdRJ3m8RC2kDhEtIiQPKg", "endpoints": ["CONTROLLER://localhost:9093"]}, {"id": 102, "directoryId": "NAtqshP9S760aRSGrKw9GA", "endpoints": ["CONTROL
CurrentObservers:       [{"id": 1, "directoryId": "OwKvqSLAVEgnDd_0Rh5V3Q"}]
leader controller: 101
=== quorum replication (voters and the observer)
NodeId	LogEndOffset	Lag	Status  
101   	86          	0  	Leader  
102   	85          	1  	Follower
103   	85          	1  	Follower
1     	86          	0  	Observer
=== create topic orders
Created topic orders.
=== 'Replayed TopicRecord' on every controller (active and standby)
node 101: INFO [QuorumController id=101] Replayed TopicRecord for topic orders with topic ID lVgPP2VQS7uVa6RLnyF7uQ. (or
node 102: INFO [QuorumController id=102] Replayed TopicRecord for topic orders with topic ID lVgPP2VQS7uVa6RLnyF7uQ. (or
node 103: INFO [QuorumController id=103] Replayed TopicRecord for topic orders with topic ID lVgPP2VQS7uVa6RLnyF7uQ. (or
=== elections so far (CandidateState transitions in all controller logs): 1
=== kill -9 the leader controller (101, epoch 1) at 05:03:33
LeaderId:               102
LeaderEpoch:            2
HighWatermark:          146
=== state transitions logged by node 102 after the kill (new leader: 102)
05:03:34,908  FollowerState -> ProspectiveState  epoch=1
05:03:35,022  ProspectiveState -> CandidateState  epoch=2
05:03:35,135  CandidateState -> Leader  epoch=2
=== state transitions logged by node 103 after the kill (new leader: 102)
05:03:34,907  FollowerState -> ProspectiveState  epoch=1
05:03:35,077  ProspectiveState -> UnattachedState  epoch=2
05:03:35,104  UnattachedState -> UnattachedState  epoch=2
05:03:35,227  UnattachedState -> FollowerState  epoch=2
=== quorum-state files of the two survivors
node 102: {"leaderId":102,"leaderEpoch":2,"votedId":-1,"votedDirectoryId":"AAAAAAAAAAAAAAAAAAAAAA","data_version":1}
node 103: {"leaderId":102,"leaderEpoch":2,"votedId":102,"votedDirectoryId":"NAtqshP9S760aRSGrKw9GA","data_version":1}
=== the cluster still works with 2 of 3 controllers
Created topic payments.

(Times, ids, offsets and which node becomes leader differ on every run. The script truncates long lines. The state transition lines keep only the time, state names and epoch from the Completed transition to … lines in server.log.)

1. Lag in --replication is how many records each voter is behind the leader. Broker 1 shows up as Observer.
2. The Replayed TopicRecord line appears on all three controllers. Only the active controller 101 created the records, but the standby controllers 102 and 103 also receive the committed records and build the same in-memory state (handleCommit() in section 3).
3. Less than two seconds after 101 was killed (the script prints the kill time to the second only), at 05:03:34.9, 102 and 103 became Prospective almost at the same moment because of the fetch timeout. 102 became Candidate (epoch 2) first and sent Vote; 103 received a request with a larger epoch, moved to Unattached (epoch 2) (lines 896–897 above), and changed once more to Unattached (voted) when it granted the vote (lines 931–933). With its own vote and 103's, 102 had two of three and became Leader; 103 got BeginQuorumEpoch and became Follower. From Prospective to a new leader took 0.23 seconds; counted from the kill, about two seconds.
4. In its quorum-state file, 103 remembers "votedId":102 — the vote durableTransitionTo() left on disk.
5. With only two controllers left (two of three = a majority), creating a topic still works.

One election, drawn with the same node numbers as Example 2. In the real run 103 also became Prospective at the same time; the picture only shows 102 sending the PreVote.

Easy to confuse: an election does not always finish in one round

I killed the leader at epoch 1 in three runs, and in the two runs before the one shown above, the new leader came out at epoch 3. In those runs the new leader's log had ProspectiveState -> CandidateState epoch=2 followed by CandidateState -> ProspectiveState epoch=2, then -> CandidateState epoch=3 → -> Leader epoch=3. The epoch-2 election did not get a majority in time (Candidate → Prospective: election timeout · loss in the diagram), so it ran again. Whether it failed because both nodes became candidates at once and split the vote, I could not confirm: those runs did not keep the other node's log (the script prints both survivors' logs from the last run on). The election timeout is controller.quorum.election.timeout.ms (default 1,000ms) plus a random amount between 0 and that value, different per node (electionTimeoutMs + random.nextInt(electionTimeoutMs), QuorumState.java lines 745–749), so in the next round one side usually moves first and wins. An epoch that jumps by two is nothing to worry about.

One more thing: the first time I ran it, several other containers were running on the same PC and the machine was very slow; the epoch had already climbed to 6 before I killed anything (and describe --replication failed with a timeout). With a fetch timeout of only 2 seconds, a JVM that stalls for a few seconds makes a healthy leader look dead. So I changed the script to ask for the leader and epoch again right before the kill and to print the number of elections so far (elections so far).

3. QuorumController — one thread on top of an event queue

One queue, one thread

In section 4 of Part 2 we saw the javadoc saying "the QuorumController is single-threaded". That thread is created and started in the constructor of KafkaEventQueue in server-common (new KafkaThread(threadNamePrefix + EVENT_HANDLER_THREAD_SUFFIX, …), KafkaEventQueue.java lines 483–488). EVENT_HANDLER_THREAD_SUFFIX is "event-handler" (line 43), and ControllerServer passes the prefix quorum-controller-<node id>- (ControllerServer.scala line 242), which gives the quorum-controller-1-event-handler from the Part 2 example. The queue is not controller-only: the broker's MetadataLoader (…metadata-loader-event-handler), BrokerLifecycleManager and the snapshot-writing SnapshotGenerator each have one thread of their own from the same class. Nearly all of KRaft follows the same pattern: "only one thread touches the state; other threads put events in".

The middle of the handleEvents() loop that thread runs:

                    if (head.next == head) {
                        if (deadlineMap.isEmpty() && (shuttingDown || interrupted)) {
                            // If there are no more entries to process, and the queue is
                            // closing, exit the thread.
                            return;
                        }
                    } else {
                        if (interrupted) {
                            toDeliver = new InterruptedException("The event handler thread is interrupted");
                        } else {
                            toDeliver = null;
                        }
                        toRun = head.next;
                        remove(toRun);
                        continue;
                    }

server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java lines 266–281

It takes the head of the linked list (head.next) into toRun and loops; at the top of the loop (lines 216–218) the event runs with the lock released. Only adding to and removing from the queue is guarded by the lock; the events themselves run one at a time on one thread, so the controller's state (the topic list, broker registrations …) has no locks at all. The part above this excerpt (lines 231–265) hands a TimeoutException to events past their deadline and pulls out deferred events when their time comes. The idea of one thread handling events in turn instead of using locks resembles the event loop in the nginx series.

ControllerResult = list of records + response

Every write operation on the controller returns a value of the same shape.

class ControllerResult<T> {
    private final List<ApiMessageAndVersion> records;
    private final T response;
    private final boolean isAtomic;

metadata/src/main/java/org/apache/kafka/controller/ControllerResult.java lines 27–30

ReplicationControlManager.createTopics() does not "create" the topic. It only computes what must be written to the log to create it (the record list) and what to answer the client once that's done (the response). What actually changes state is a separate replay(record). Because "deciding" and "applying" are split like this, standby controllers and brokers can apply exactly what the active controller decided just by receiving the records. isAtomic asks for the records to go into one batch so that they are written all together or not at all.

The life of one write event

appendWriteEvent() just builds a ControllerWriteEvent, puts it in the queue and returns its future (lines 952–965). The real work happens in the event's run() once it is taken off the queue.

            ControllerResult<T> result = op.generateRecordsAndResult();
            if (result.records().isEmpty()) {
                // ...
            } else {
                // Pass the records to the Raft layer. This will start the process of committing
                // them to the log.
                long offset = appendRecords(log, result, maxRecordsPerBatch,
                    records -> {
                        // Start by trying to apply the record to our in-memory state. This should always
                        // succeed; if it does not, that's a fatal error. It is important to do this before
                        // scheduling the record for Raft replication.
                        int recordIndex = 0;
                        long lastOffset = raftClient.prepareAppend(controllerEpoch, records);
                        long baseOffset = lastOffset - records.size() + 1;
                        for (ApiMessageAndVersion message : records) {
                            long recordOffset = baseOffset + recordIndex;
                            try {
                                replay(message.message(), Optional.empty(), recordOffset);
                            } catch (Throwable e) {
                                // ...
                            }
                            recordIndex++;
                        }
                        raftClient.schedulePreparedAppend();
                        offsetControl.handleScheduleAppend(lastOffset);
                        return lastOffset;
                    }
                );
                op.processBatchEndOffset(offset);
                resultAndOffset = ControllerResultAndOffset.of(offset, result);
                // ...
            }

            // Remember the latest offset and future if it is not already completed
            if (!future.isDone()) {
                deferredEventQueue.add(resultAndOffset.offset(), this);
            }

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java lines 792–853

(The omitted parts handle read-only operations that produce no records, treat a failed replay as a fatal error, and log at debug level.)

1. generateRecordsAndResult() — builds the ControllerResult (for createTopics(), a TopicRecord plus PartitionRecords).
2. raftClient.prepareAppend() — puts the records into Raft's batch accumulator (BatchAccumulator) and gets offsets back. Nothing has been written to disk or replicated yet.
3. replay() — as the comment stresses, applies the records to its own memory before they commit, so the very next event (say, creating a topic with the same name again) sees the decision just made. The Replayed TopicRecord log lines in Examples 1 and 2 come from this replay().
4. schedulePreparedAppend() and handleScheduleAppend(lastOffset) — tell Raft "you can write these now" and take an in-memory snapshot at that offset (section 4).
5. deferredEventQueue.add(offset, this) — the future is not completed yet; the event waits in line for "wake me when offset offset commits". This is the "the future does not complete until the result is durable in the metadata log" we saw in Part 2.

Three threads and one other voter share a single write. The 25ms in step 6 is the default of controller.quorum.append.linger.ms (DEFAULT_QUORUM_LINGER_MS = 25, QuorumConfig.java line 94) — like the producer's linger.ms (Part 5), it gathers a little before writing.

When the commit arrives: the active controller wakes futures, standbys replay

Here is QuorumMetaLogListener.handleCommit(), called from pollListeners() in KafkaRaftClient.

        public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
            appendRaftEvent("handleCommit[baseOffset=" + reader.baseOffset() + "]", () -> {
                try {
                    boolean isActive = isActiveController();
                    while (reader.hasNext()) {
                        Batch<ApiMessageAndVersion> batch = reader.next();
                        long offset = batch.lastOffset();
                        int epoch = batch.epoch();
                        List<ApiMessageAndVersion> messages = batch.records();

                        if (messages.isEmpty()) {
                            // ...
                        } else if (isActive) {
                            // If the controller is active, the records were already replayed,
                            // so we don't need to do it here.
                            log.debug("Completing purgatory items up to offset {} and epoch {}.", offset, epoch);

                            // Advance the committed and stable offsets then complete any pending purgatory
                            // items that were waiting for these offsets.
                            offsetControl.handleCommitBatch(batch);
                            deferredEventQueue.completeUpTo(offsetControl.lastStableOffset());
                        } else {
                            // If the controller is a standby, replay the records that were
                            // created by the active controller.
                            // ... (replay(message.message(), Optional.empty(), recordOffset) for each record)
                            offsetControl.handleCommitBatch(batch);
                        }
                    }
                } finally {
                    reader.close();
                }
            });
        }

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java lines 969–1020

1. appendRaftEvent(...) — handleCommit() is called on raft-io-thread, but it does not do the work there; it puts it on the controller's event queue, because only the event thread touches the controller's state.
2. The active controller has already run replay(), so it doesn't again; it advances the committed offset and completes, with completeUpTo(), the futures of all write events waiting up to that offset. That is when ControllerApis sends the response to the client.
3. A standby controller runs replay() on the same records here. A standby only applies committed records, so it never has to undo anything. This is the path that printed Replayed TopicRecord on 102 and 103 in Example 2.

replay() is a switch that sends each record type to the manager responsible for it (like case TOPIC_RECORD: replicationControl.replay((TopicRecord) message);, QuorumController.java lines 1240–1262). There is one manager per area — ClusterControlManager for broker registrations, ReplicationControlManager for topics and partitions, ConfigurationControlManager for configs, FeatureControlManager for feature levels …

4. Memory you can roll back — SnapshotRegistry and TimelineHashMap

Applied first, but what if it never commits?

In section 3 the active controller applied records to memory before they committed. What if this controller loses leadership before the commit (say, the network splits and another node wins with a larger epoch)? The records may not be in the new leader's log and get truncated to match it. Then whatever was applied to memory early has to be undone as well.

That is why the controller's state structures are TimelineHashMap, not plain HashMap.

    /**
     * Maps topic names to topic UUIDs.
     */
    private final TimelineHashMap<String, Uuid> topicsByName;
    // ...
    /**
     * Maps topic UUIDs to structures containing topic information, including partitions.
     */
    private final TimelineHashMap<Uuid, TopicControlInfo> topics;

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java lines 336–361

replay(TopicRecord) simply puts into these maps (topicsByName.put(record.name(), record.topicId()), line 427; topics.put(record.topicId(), new TopicControlInfo(…)), lines 449–450). Each map is given the shared SnapshotRegistry when it is created (lines 416–418) and registers itself there. The description of the base class of the timeline package (SnapshottableHashTable.java lines 26–31) sums up the approach. Taking a snapshot (an in-memory one here) does not copy the whole map. When an entry is overwritten or removed after a snapshot, only a reference to the old value is moved into the most recent snapshot tier at that moment. So frequent snapshots cost memory only for what changed. Rolling back puts the saved old values back in place. This "in-memory snapshot" shares only its name with the on-disk snapshot (.checkpoint) in section 5.

When snapshots are taken, dropped and reverted to

QuorumController takes these snapshots by offset, and OffsetControlManager decides when. handleScheduleAppend(lastOffset), called after the records are applied, takes a snapshot at that offset with snapshotRegistry.idempotentCreateSnapshot(lastOffset) (OffsetControlManager.java lines 302–305), and when a commit arrives it does this:

    void maybeAdvanceLastStableOffset() {
        long newLastStableOffset;
        if (transactionStartOffset == -1L) {
            newLastStableOffset = lastCommittedOffset;
        } else {
            newLastStableOffset = Math.min(transactionStartOffset - 1, lastCommittedOffset);
        }
        if (lastStableOffset < newLastStableOffset) {
            lastStableOffset = newLastStableOffset;
            snapshotRegistry.deleteSnapshotsUpTo(lastStableOffset);
            if (!active()) {
                snapshotRegistry.idempotentCreateSnapshot(lastStableOffset);
            }
        }
    }

metadata/src/main/java/org/apache/kafka/controller/OffsetControlManager.java lines 319–333

1. Right before becoming active (activate()), it takes a snapshot at the last committed offset (lines 249–252).
2. Every time records are applied, it takes a snapshot at their last offset (handleScheduleAppend).
3. When a commit arrives, older snapshots are no longer needed and are dropped (deleteSnapshotsUpTo).
4. On losing leadership (renounce() → deactivate(), QuorumController.java lines 1204–1216) it reverts to the snapshot at the last committed offset (snapshotRegistry.revertToSnapshot(lastStableOffset), OffsetControlManager.java lines 260–272). At the same time deferredEventQueue.failAll(...) ends the waiting futures with a "not the controller" error, so clients retry against the new controller.

(transactionStartOffset is for metadata transactions, groups of records that span several batches. The BEGIN_TRANSACTION_RECORD and END_TRANSACTION_RECORD in Example 1 wrap the bootstrap records (feature levels and so on) that the first active controller writes as one group (new BeginTransactionRecord().setName("Bootstrap records"), ActivationRecordsGenerator.java line 81) — not the producer transactions of Part 12.)

Example 3: roll back a TimelineHashMap yourself

No server this time. We load SnapshotRegistry and TimelineHashMap straight from libs/kafka-server-common-4.5.0-SNAPSHOT.jar in the distribution and act out steps 1–4 above in the same order (ex/u10-timeline/Timeline.java).

// An example written for this post (not part of Kafka's source)
// Imitates, in the same order as QuorumController and OffsetControlManager,
// how the active controller applies metadata records to memory "before they commit"
// and rolls back to the last committed offset when it loses leadership.
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
// ...
public class Timeline {
    static SnapshotRegistry registry = new SnapshotRegistry(new LogContext());
    // a "topic name -> topic id" map, like ReplicationControlManager.topicsByName
    static TimelineHashMap<String, String> topicsByName = new TimelineHashMap<>(registry, 0);

    public static void main(String[] args) {
        // say everything up to offset 10 is already committed
        topicsByName.put("orders", "id-A");
        // activate(): take a snapshot at the last committed offset (10) before becoming active
        registry.idempotentCreateSnapshot(10);
        print("committed up to 10");

        // record 11: apply TopicRecord(payments), then snapshot 11 like handleScheduleAppend(11)
        topicsByName.put("payments", "id-B");
        registry.idempotentCreateSnapshot(11);
        // record 12: apply RemoveTopicRecord(orders), then snapshot 12
        topicsByName.remove("orders");
        registry.idempotentCreateSnapshot(12);
        print("applied 11, 12 (not committed)");

        // past values can be read too — get(key, epoch)
        System.out.println("  as of 10: orders=" + topicsByName.get("orders", 10)
            + " payments=" + topicsByName.get("payments", 10));
        System.out.println("  as of 11: orders=" + topicsByName.get("orders", 11)
            + " payments=" + topicsByName.get("payments", 11));

        // Raft committed up to 11 -> maybeAdvanceLastStableOffset(): drop snapshots older than 11
        registry.deleteSnapshotsUpTo(11);
        print("committed up to 11");

        // leadership lost before 12 committed -> deactivate(): revert to the last committed offset (11)
        registry.revertToSnapshot(11);
        print("lost leadership, reverted to 11");
    }
    // ... print(): prints the map's current content and registry.epochsList() on one line
}

=== jar that holds org.apache.kafka.timeline
kafka-server-common-4.5.0-SNAPSHOT.jar
=== Timeline.java
committed up to 10                 current={orders=id-A} snapshots=[10]
applied 11, 12 (not committed)     current={payments=id-B} snapshots=[10, 11, 12]
  as of 10: orders=id-A payments=null
  as of 11: orders=id-A payments=id-B
committed up to 11                 current={payments=id-B} snapshots=[11, 12]
lost leadership, reverted to 11    current={orders=id-A, payments=id-B} snapshots=[11]

1. Even after record 12 (removing orders) is applied, get("orders", 11) returns id-A, because a reference to the old value was moved into the snapshot-11 tier. The controller uses this to answer read requests such as describeConfigs with values as of the committed point (configurationControl.describeConfigs(offsetControl.lastStableOffset(), resources), QuorumController.java line 1884).
2. Once 11 committed, snapshot 10 disappeared (deleteSnapshotsUpTo(11) drops only snapshots smaller than 11, SnapshotRegistry.java lines 297–305).
3. Reverting to 11 after losing leadership removed only the effect of uncommitted record 12 (deleting orders); the effect of committed record 11 (payments) stayed.

The path Example 3 takes. Green is committed state, yellow is applied but not committed, blue is what happens when a commit arrives, red is the rollback on losing leadership.

Easy to confuse: snapshot N is the state "after applying records up to N"

handleScheduleAppend(lastOffset) is called after the records are replay()ed (see the order in the section 3 excerpt). So snapshot 11 is not "before applying record 11" but "after applying records up to 11", and revertToSnapshot(lastStableOffset) returns to "the state with every committed record applied". Example 3 likewise calls idempotentCreateSnapshot(11) after put("payments"). Do it the other way round and a rollback also throws away one committed record.

5. How brokers read the ledger — MetadataLoader, MetadataDelta, MetadataImage

Brokers turn records into an "image"

The MetadataLoader that SharedServer creates is another Raft listener. Its class description says it "follows changes provided by a RaftClient and packages them into metadata deltas and images that publishers can consume, calls publishers from its own single thread in the order they were installed, and publishers must not modify the data structures passed to them" (MetadataLoader.java lines 63–70). Two new words:

• MetadataImage: an immutable object holding the whole ledger up to some offset. It is a Java record class whose fields are the features, cluster (brokers), topics, configs, client quotas, producer ids, ACLs, SCRAM and delegation token images (MetadataImage.java lines 31–34). Once built it never changes, so many threads can read it without locks. The KRaftMetadataCache we saw in Part 2 reads this image when answering Metadata requests.
• MetadataDelta: the difference produced by applying records to the previous image. Each replay(record) creates and accumulates only the delta for the area the record touches (case TOPIC_RECORD: replay((TopicRecord) record); → getOrCreateTopicsDelta().replay(record), MetadataDelta.java lines 195–209, lines 287–289).

When a committed batch arrives, the loader does this on its own event thread:

    @Override
    public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
        eventQueue.append(() -> {
            try (reader) {
                while (reader.hasNext()) {
                    Batch<ApiMessageAndVersion> batch = reader.next();
                    loadControlRecords(batch);
                    long elapsedNs = batchLoader.loadBatch(batch, currentLeaderAndEpoch);
                    // ...
                }
                batchLoader.maybeFlushBatches(currentLeaderAndEpoch, true);
            } catch (Throwable e) {
                // ...
            }
        });
    }

metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java lines 395–414

batchLoader.loadBatch() replays the batch's records into the delta, and maybeFlushBatches() builds a new image from the accumulated delta and hands it to the publishers (publisher.onMetadataUpdate(delta, image, manifest) in maybePublishMetadata(), lines 362–370). MetadataDelta.apply(), which builds the new image, is a row of branches shaped like if (featuresDelta == null) { newFeatures = image.features(); } else { newFeatures = featuresDelta.apply(); }, one per area (MetadataDelta.java lines 395–413 onward). An area with no delta reuses the previous image's object as-is. Creating one topic does not rebuild the cluster, config or ACL images; they are shared with the previous image. Because the objects are immutable, sharing them is safe.

Publishers receive both the delta (what changed) and the new image (what everything looks like now). How ReplicaManager.applyDelta() picks out of the delta only "changes to partitions that include my node" and takes on leader or follower roles was covered in section 5 of Part 2.

On a broker, BrokerServer installs the publishers. There are also publishers for configs (DynamicConfigPublisher), quotas (DynamicClientQuotaPublisher), SCRAM, delegation tokens and ACLs (AclPublisher), but they are not installed on the loader separately: they are handed to BrokerMetadataPublisher when it is built (BrokerServer.scala lines 531–576), and BrokerMetadataPublisher.onMetadataUpdate() calls them one after another (BrokerMetadataPublisher.scala lines 189–205). Change a topic config with kafka-configs.sh and a ConfigRecord is written; it reaches DynamicConfigPublisher on every broker and the actual setting changes.

No announcements until it has caught up

If a freshly started broker read the log from the start and notified publishers for every record, ReplicaManager would live through "topic created, deleted, created again" one step at a time. So the loader doesn't call publishers until it first catches up to the high watermark. stillNeedToCatchUp() decides "still catching up if the offset read so far is below high watermark − 1" (MetadataLoader.java lines 260–264), and maybePublishMetadata() only swaps the image and returns while that check is true (lines 352–357). Once caught up, it rolls every change so far into one delta and announces it in one go.

Snapshots — so nobody reads the log from the start

As the metadata log keeps growing, a newly added broker or a controller that was down for a long time would have to replay millions of records from the start. So a snapshot is made: the whole image up to some offset written out as records (a <end offset>-<epoch>.checkpoint file). The snapshot writer is another MetadataLoader publisher, SnapshotGenerator (installed by SharedServer, SharedServer.scala lines 346–356).

publishLogDelta() (SnapshotGenerator.java lines 232–253) writes the current image to a file once the bytes applied since the last snapshot exceed metadata.log.max.record.bytes.between.snapshots (default 20MB, MetadataLogConfig.java lines 40–41) or the time exceeds metadata.log.max.snapshot.interval.ms (default 1 hour, lines 38–39). Once a snapshot exists, log segments before it can be deleted. To a follower fetching with an empty log (when a non-bootstrap snapshot exists, lines 1625–1628) or one that has fallen behind the start of the leader's log, the leader answers "take this snapshot" instead of records, and the follower downloads it with FetchSnapshot — the fifth API in the javadoc of section 2. Example 1 ran only briefly, so it had just the bootstrap snapshot from formatting.

Example 1, continued: browsing the image with kafka-metadata-shell.sh

kafka-metadata-shell.sh reads a metadata log (or snapshot) file and shows the image like a file system. It locks the folder the server uses, so I ran it after stopping the server.

=== metadata shell over the log file (after stop)
--- ls /image
acls clientQuotas cluster configs delegationToken features producerIds provenance scram topics 
--- ls /image/topics/byName/orders
0 1 2 id name 
--- cat /image/topics/byName/orders/0
PartitionRegistration(replicas=[1], directories=[5H7XVkyAS9mZZQhe8CMRaw], isr=[], removingReplicas=[], addingReplicas=[], elr=[1], lastKnownElr=[1], leader=-1, leaderRecoveryState=RECOVERED, leaderEpo
--- cat /image/provenance
offset 71, epoch 1, time 2026-11-07T04:55:37.811Z[UTC]

The names under /image are exactly the field names of MetadataImage. The tool doesn't imitate the image; it builds the same MetadataLoader as the broker and feeds it the file (shell/src/main/java/org/apache/kafka/shell/MetadataShell.java lines 154–161). So what you see here is the same image the broker had in memory.

Easy to confuse: a log read after shutdown includes the shutdown itself

In the output above, orders-0 shows isr=[] and leader=-1, which caught my eye: the record at topic creation had isr=[1] and leader=1. The shell read to the end of the log (offset 71) after the server was stopped, and in between, broker 1 did a controlled shutdown and the controller wrote a record saying "remove broker 1 from the ISR" (handleBrokerInControlledShutdown() calls generateLeaderAndIsrUpdates(...) for every ISR the broker is in, ReplicationControlManager.java lines 1514–1523). With only one replica the ISR became empty, and the broker was kept in elr=[1] instead. ELR (Eligible Leader Replicas) is the list of replicas that left the ISR but are still eligible to become leader again; the comment in PartitionRecord.json names it as KIP-966 (the eligible.leader.replicas.version feature was at level 1 in Example 1). ELR rules are not covered in this post. The metadata log is not "the current state" but "everything that happened", so always check how far it was read (/image/provenance).

6. Broker registration, heartbeats and fencing

The broker side: BrokerLifecycleManager

Part 2 said a broker only starts serving clients after it "is recognised as caught up on metadata and the fence is lifted". BrokerLifecycleManager does that. On startup it registers itself with a BrokerRegistration request (the controller writes a RegisterBrokerRecord), and from then on it sends heartbeats.

    private void sendBrokerHeartbeat() {
        Long metadataOffset = highestMetadataOffsetProvider.get();
        BrokerHeartbeatRequestData data = new BrokerHeartbeatRequestData()
            .setBrokerEpoch(brokerEpoch)
            .setBrokerId(nodeId)
            .setCurrentMetadataOffset(metadataOffset)
            .setWantFence(!readyToUnfence)
            .setWantShutDown(state == BrokerState.PENDING_CONTROLLED_SHUTDOWN)
            .setOfflineLogDirs(new ArrayList<>(offlineDirs.keySet()));

server/src/main/java/org/apache/kafka/server/BrokerLifecycleManager.java lines 564–572

A heartbeat is both an "I'm alive" signal and a report. currentMetadataOffset is how far this broker's MetadataLoader has read, wantFence means "I'm not ready yet, keep the fence up", and wantShutDown means "I'd like to go down now". The interval is broker.heartbeat.interval.ms (default 2,000ms, KRaftConfigs.java lines 39–44).

The controller side: caught up, fenced, session expired

The controller handles a heartbeat in ReplicationControlManager.processBrokerHeartbeat() and decides the broker's next state. Whether the broker has caught up comes down to this line:

        boolean isCaughtUp = request.currentMetadataOffset() >= registerBrokerRecordOffset;
        BrokerHeartbeatReply reply = new BrokerHeartbeatReply(isCaughtUp,
                states.next().fenced(),
                states.next().inControlledShutdown(),
                states.next().shouldShutDown());
        return ControllerResult.of(records, reply);

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java lines 1753–1758

The broker counts as caught up once the offset it has read reaches the offset of its own registration record. That means it has read the ledger at least up to "I was registered", and so it knows every topic and partition before that. Notice that heartbeat handling is also a write event returning a ControllerResult: when the state changes (the fence comes down), records gets a record.

Brokers whose heartbeats stop are found by a periodic task. The controller runs a task called maybeFenceStaleBroker every 1/8 of the session timeout (QuorumController.java lines 1684–1701), and this is what it calls:

        // Even though multiple brokers can go stale at a time, we will process
        // fencing one at a time so that the effect of fencing each broker is visible
        // to the system prior to processing the next one.
        log.info("Fencing broker {} at epoch {} because its session has timed out.", id, epoch);
        List<ApiMessageAndVersion> records = new ArrayList<>();
        handleBrokerFenced(id, records);
        heartbeatManager.fence(id);
        return ControllerResult.of(records, true);

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java lines 1809–1816

handleBrokerFenced() (lines 1449–1460) produces two kinds of records: first, for every partition whose ISR contains the broker, a PartitionChangeRecord that removes it from the ISR and changes the leader if needed (generateLeaderAndIsrUpdates(...)); then a BrokerRegistrationChangeRecord saying the broker is fenced (setFenced(BrokerRegistrationFencingChange.FENCE.value())). The session length is broker.session.timeout.ms (default 9,000ms).

The state the broker keeps for itself (BrokerState). Only the move to PENDING_CONTROLLED_SHUTDOWN on a shutdown signal is the broker's own decision (line 343); every other condition is a value the controller puts in the heartbeat reply (the switch at BrokerLifecycleManager.java lines 645–702). The other direction, the controller fencing a broker, happens separately, outside this picture.

Example 4: freeze a broker and watch it get fenced

We start with the shared helper kafka_cluster3 (controller node 100 plus brokers 1, 2, 3), create a topic with replication factor 3, and freeze broker 3 with kill -STOP. The process is still alive but can't do anything, so its heartbeats stop. After 15 seconds we look at the controller log and the metadata log, then thaw it with kill -CONT and look again (ex/u10-fence/run.sh).

=== broker lifecycle log of broker 3 (startup)
INFO [BrokerLifecycleManager id=3] The broker has caught up. Transitioning from STARTING to RECOVERY. (org.apache.kafka.
INFO [BrokerLifecycleManager id=3] The broker has been unfenced. Transitioning from RECOVERY to RUNNING. (org.apache.kaf
=== create topic orders (1 partition, RF=3)
 Topic: orders Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr: 
=== freeze broker 3 (kill -STOP) at 04:58:24
=== controller log
04:58:31 INFO [QuorumController id=100] Fencing broker 3 at epoch 57 because its session has timed out. (org.apache.kafka.controller.Replica
 Topic: orders Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2 Elr: LastKnownElr: 
=== new metadata records after offset 113 (NoOp records omitted)
offset 131 {"type":"PARTITION_CHANGE_RECORD","version":2,"data":{"partitionId":0,"topicId":"1YXyxrGoS_uwbCoz0XEUeA","isr":[1,2]}}
offset 132 {"type":"BROKER_REGISTRATION_CHANGE_RECORD","version":0,"data":{"brokerId":3,"brokerEpoch":57,"fenced":1}}
=== thaw broker 3 (kill -CONT)
=== new metadata records after offset 163 (NoOp records omitted)
offset 164 {"type":"BROKER_REGISTRATION_CHANGE_RECORD","version":0,"data":{"brokerId":3,"brokerEpoch":57,"fenced":-1}}
offset 166 {"type":"PARTITION_CHANGE_RECORD","version":2,"data":{"partitionId":0,"topicId":"1YXyxrGoS_uwbCoz0XEUeA","isr":[1,2,3]}}
 Topic: orders Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr: 

(Times, broker epochs, offsets, topic ids and which broker starts as leader differ on every run.)

1. The two startup lines are exactly STARTING → RECOVERY → RUNNING from the diagram — the logger.info messages at lines 648 and 666, printed verbatim.
2. Seven seconds after freezing (04:58:24 → 04:58:31) the controller fenced broker 3. The last heartbeat went out sometime in the 2-second interval before the freeze, and the check runs every 1.125 seconds (1/8 of 9 seconds), so "9 seconds since the last heartbeat" comes out at roughly 7–10 seconds after the freeze (the printed times drop the fractions of a second). Other runs gave 10 and 9 seconds. Once, run at the same time as three other examples on a busy machine, it was 6 seconds; I did not find out why.
3. In the metadata log, in the same order handleBrokerFenced() builds them, the PARTITION_CHANGE_RECORD removing 3 from the ISR is followed by fenced: 1. In this run the leader was broker 1, so only isr changed; in another run where broker 3 was the leader, the same record also carried "leader":1 and leadership moved.
4. After thawing, broker 3's heartbeats arrived again and fenced: -1 was written (fence lifted, handleBrokerUnfenced() lines 1494–1501); a moment later the ISR was back to 1, 2, 3. handleBrokerUnfenced() only touches partitions without a leader (brokersToIsrs.partitionsWithNoLeader(), lines 1499–1500), so the ISR coming back for this partition, which had a leader, is not the unfencing itself: it is the ISR change request the leader (broker 1) sent saying "3 has caught up" (AlterPartition, Part 9), turned into a record.

Easy to confuse: there are two ways out of the ISR

When a broker stops responding, it can leave the ISR in two ways. One is the controller's fencing in this section (no heartbeat for broker.session.timeout.ms; all partitions the broker is in, at once). The other is the leader shrinking the ISR from Part 9 (the follower's Fetch can't keep up for replica.lag.time.max.ms; that partition only). In Example 4 the 9-second fencing happened first, and the Fencing broker 3 … session has timed out line tells you which path it was. When you wonder why an ISR shrank, look at the controller's controller.log and the leader broker's server.log together.

What this post leaves out

• Adding and removing voters at run time (dynamic quorum, AddRaftVoter, RemoveRaftVoter, UpdateRaftVoter). raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java, RemoveVoterHandler.java, UpdateVoterHandler.java, KRaftControlRecordStateMachine.java
• The details of follower log truncation (diverging epoch) and snapshot download. KafkaRaftClient.handleFetchResponse(), handleFetchSnapshotResponse(), raft/src/main/java/org/apache/kafka/snapshot/
• Check quorum, where a leader that doesn't hear Fetches from a majority steps down. LeaderState.timeUntilCheckQuorumExpires()
• Replica placement (which brokers get replicas), preferred and unclean leader election, ELR. metadata/src/main/java/org/apache/kafka/metadata/placement/, PartitionChangeBuilder.java, electLeaders in ReplicationControlManager
• Feature levels (metadata.version etc.) and upgrades. FeatureControlManager.java, server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
• Metadata transactions (BeginTransactionRecord). OffsetControlManager.replay(BeginTransactionRecord, …), ActivationRecordsGenerator.java
• The managers and publishers for configs, quotas, ACLs, SCRAM and delegation tokens. ConfigurationControlManager, ClientQuotaControlManager, AclControlManager, ScramControlManager, DelegationTokenControlManager
• Formatting and the bootstrap snapshot. core/src/main/scala/kafka/tools/StorageTool.scala, metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java
• Left out of the whole series: Kafka Streams, Kafka Connect, MirrorMaker, share groups, tiered storage, security (SASL, TLS, ACL authorization), quotas, the Admin client, JBOD

Summary

What's next

The next post, Part 11: The Consumer and the Group Coordinator — Inside poll() and Rebalancing, moves to another state machine that keeps the cluster running: the group coordinator, which holds consumer group members, assignments and committed offsets. Interestingly, its structure resembles this post. The group coordinator also writes its state as records to a log (__consumer_offsets), reads them back to rebuild in-memory state, and uses a runtime that handles events one at a time per partition (CoordinatorRuntime) along with this post's TimelineHashMap. We'll see how the "the log is the state" idea is reused outside the controller.

Related posts:

• Part 2: The Big Picture — region G zoomed in here, and the full flow of creating one topic
• Part 9: Replication — The ISR, the High Watermark and the Wait Behind acks=all — data partition replication over the same Fetch, and an ISR-based high watermark
• Part 7: Log Storage — the segments and indexes the metadata log uses too
• Digging into nginx's Source — another design where one thread handles events in turn