02 - KRaft in Kafka 4.0: Goodbye ZooKeeper, Quorum Controller and Migration

Who decides where each partition lives, now that ZooKeeper is gone

A Kafka cluster must know at every moment which brokers are alive, who is the leader of each partition, which topics exist and which ACLs apply. Up to the 3.x series this registry lived in an external system, Apache ZooKeeper, to be installed, monitored and upgraded separately. Kafka 4.0 removed that dependency: the upgrade documentation says unambiguously that "Kafka 4.0 only supports KRaft mode - ZooKeeper mode has been removed". If you still have a cluster with ZooKeeper, to move to 4.x you must first migrate it, and the path goes through a bridge release that is not the one many people remember: 3.9.

In this article you understand how KRaft keeps the metadata, how the controller quorum is sized, how a cluster is configured from scratch and how the migration is carried out. All the configuration facts come from the Kafka 4.4 documentation (the "KRaft" page) and from the 3.9 one for the migration, consulted on 2026-11-12; the behaviours of the consensus are demonstrated by a small Java 21 model that I compiled and ran. The model is not Kafka: it exchanges no network messages, has no growing epochs and no snapshots. It serves to make one rule visible, the one by which a confirmed record survives the death of the leader.

Why remove ZooKeeper: the reason according to those who wrote the proposal

The original proposal, KIP-500, motivates the choice with two concrete arguments. The first is state divergence: the controller kept a copy of the metadata in memory and compared it with ZooKeeper through "watches", which signal that something has changed but not what. The second is the cost of changing controller: whoever took over had to reload the state. In KRaft the controllers all follow the same log, so whoever takes over already has the state (the KIP speaks of "hot standbys"). The same proposal adds a promise of scalability (more partitions with less CPU), but it is a project goal, not a measurement: no figure for maximum partitions or failover seconds is reported here, because I have no source that demonstrates it.

If you have read precise figures elsewhere (a maximum number of partitions for ZooKeeper, the seconds of a failover, a percentage reduction quoted from a company case), treat them as unverified: they do not appear in the official documentation, which is why I do not repeat them. The operational advantage you can state is instead structural: one system to manage instead of two, a single upgrade cycle, and the cluster state contained in a single log instead of scattered between Kafka and ZooKeeper.

How KRaft works: the metadata is a log, and the brokers read it

The core of KRaft is an idea you have already met in the article on the fundamentals: if data is a log, coordination is a log. The cluster metadata (topics, partitions, leaders, ISR, ACLs, configurations) is written as events in a special partition, __cluster_metadata-0, replicated among the nodes with the controller role through the Raft consensus protocol, described in the paper "In Search of an Understandable Consensus Algorithm". Among the controllers only one is elected, the active controller, which receives the brokers' requests and writes the events; the others replicate and stay ready to take over.

A detail that is often misunderstood: brokers do not receive the metadata by push. According to KIP-500 the brokers fetch it from the active controller (with a request similar to the consumers' fetch), keep track of the last offset received and ask only for what is new. That is why a broker that restarts does not have to reload the whole state. The events arrive in the same order on all nodes, because that is the order of the log.

The rule that makes the consensus safe is the majority. An event is committed when it is in the log of more than half of the controllers; a candidate leader obtains a node's vote only if its log is at least as up to date as the voter's. The following model applies exactly these two rules, in two scenes: in the first one a controller has fallen behind, in the second the last event exists only on the leader that fails.

import java.util.*;

public class MetadataLog {
    record Entry(int epoch, String event) {}

    static class Node {
        final String name; final List<Entry> log = new ArrayList<>(); boolean alive = true;
        Node(String name) { this.name = name; }
        int lastEpoch() { return log.isEmpty() ? 0 : log.get(log.size() - 1).epoch(); }
        // Raft voting rule: you only vote for a candidate whose log is at least as up to date as your own.
        boolean votesFor(Node candidate) {
            if (candidate.lastEpoch() != lastEpoch()) return candidate.lastEpoch() > lastEpoch();
            return candidate.log.size() >= log.size();
        }
    }

    // An entry is committed when it is in the log of a majority of the nodes.
    static int committed(List<Node> nodes) {
        int[] lengths = nodes.stream().mapToInt(n -> n.log.size()).sorted().toArray();
        return lengths[(nodes.size() - 1) / 2];  // the length reached by at least a majority
    }

    static void election(List<Node> nodes, Node candidate) {
        int votes = 1;                                      // votes for itself
        for (Node n : nodes) if (n != candidate && n.alive && n.votesFor(candidate)) votes++;
        int majority = nodes.size() / 2 + 1;
        System.out.println(candidate.name + " asks for votes: " + votes + "/" + majority
                + (votes >= majority ? " -> LEADER with " + candidate.log.size() + " entries" : " -> not elected"));
    }

    static void scene(String title, int entriesA, int entriesB, int entriesC, String... candidates) {
        System.out.println("== " + title);
        Node a = new Node("A"), b = new Node("B"), c = new Node("C");
        int[] counts = {entriesA, entriesB, entriesC};
        Node[] all = {a, b, c};
        for (int i = 0; i < 3; i++)
            for (int k = 1; k <= counts[i]; k++) all[i].log.add(new Entry(1, "event-" + k));
        List<Node> nodes = List.of(a, b, c);
        System.out.println("log A/B/C = " + entriesA + "/" + entriesB + "/" + entriesC + ", committed = " + committed(nodes));
        a.alive = false;                                    // the leader A fails
        for (String who : candidates) election(nodes, who.equals("B") ? b : c);
        int lost = a.log.size() - committed(nodes);
        System.out.println("entries present only on A and never committed (lost): " + lost);
    }

    public static void main(String[] args) {
        scene("C has fallen behind", 5, 5, 3, "C", "B");
        scene("the last entry was only on A", 6, 5, 5, "C");
    }
}

// Output:
// == C has fallen behind
// log A/B/C = 5/5/3, committed = 5
// C asks for votes: 1/2 -> not elected
// B asks for votes: 2/2 -> LEADER with 5 entries
// entries present only on A and never committed (lost): 0
// == the last entry was only on A
// log A/B/C = 6/5/5, committed = 5
// C asks for votes: 2/2 -> LEADER with 5 entries
// entries present only on A and never committed (lost): 1

In the first scene A (the leader) and B have five events, C has three; committed is 5 because two nodes out of three have them. When A fails, C asks for votes but B refuses them, because its log is shorter: C obtains only its own vote (1 out of the 2 required) and is not elected. B instead obtains C's vote and is elected with all five events. In the second scene the sixth entry was only on A: the rule says it had never been committed, and indeed the new leader starts with five entries and that one is lost. It is not a defect: no client was ever told that the entry had been applied. The model leaves out everything that in a real broker is indispensable: the epoch numbers that grow at every election, the rejection of obsolete leaders, the fetch requests with which followers align, and the snapshots that shorten the log.

How many controllers and where to put them

To tolerate N simultaneous failures you need 2N + 1 controllers, as the "Deploying Considerations" section of the documentation recalls: with 3 controllers the cluster survives one that fails, with 5 it survives two. The table below computes it for sizes from 1 to 7.

public class Quorum {
    public static void main(String[] args) {
        System.out.println("controllers  majority  failures tolerated");
        for (int n = 1; n <= 7; n++) {
            int majority = n / 2 + 1;             // more than half of the nodes
            int tolerated = n - majority;         // how many can fail while the majority stays alive
            System.out.printf("%11d  %8d  %18d%n", n, majority, tolerated);
        }
    }
}

// Output:
// controllers  majority  failures tolerated
//           1         1                   0
//           2         2                   0
//           3         2                   1
//           4         3                   1
//           5         3                   2
//           6         4                   2
//           7         4                   3

Look at rows 3 and 4: going from three to four controllers does not increase the failures tolerated (it stays 1) but requires a bigger majority, so an even number of controllers is a cost without benefit. For the same reason the sensible sizes are 3 and 5 (what the KRaft page says: "a Kafka admin will typically select 3 or 5 servers"). A single controller, row 1, tolerates zero failures: fine for development. As for memory, the documentation estimates that for a typical cluster 5 GB of RAM and 5 GB of disk for the metadata log directory are enough, an estimate of theirs ("we believe"), not a guarantee.

The next choice is between dedicated and combined controllers, those that also act as brokers (process.roles=broker,controller). The documentation says that combined ones are simpler for development, but that in combined mode the controller is less isolated and you cannot upgrade or scale the controllers separately from the brokers, and it advises against it for critical clusters. For production the road is therefore: three dedicated controllers and N brokers.

Configuring a KRaft cluster from scratch

The difference that surprises those coming from ZooKeeper is that the storage must be formatted with a cluster identifier chosen by you, before the first start. The documentation explains why: Kafka used to format empty directories automatically, and this could hide errors; if a majority of controllers could start with an empty log, a leader without already committed data could be elected. The cluster ID is generated once with kafka-storage.sh random-uuid and must be reused for every node. Keep it: you need it to add nodes.

# Dedicated controller (template to adapt, from the Kafka 4.4 documentation)
process.roles=controller
node.id=1
listeners=CONTROLLER://controller1.example.com:9093
controller.listener.names=CONTROLLER
controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093
log.dirs=/var/lib/kafka/metadata

# "Pure" broker (another file, on another machine)
# process.roles=broker
# node.id=4
# controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093
# controller.listener.names=CONTROLLER

The three properties that define the roles are process.roles, node.id (it replaces broker.id, and lives in a single number space for brokers and controllers) and controller.listener.names, the name of the listener on which the controllers talk to each other. controller.quorum.bootstrap.servers applies to every node, brokers included: it must list the controllers and serves to find the quorum, like bootstrap.servers for a client. If you get it wrong, the node does not find the active controller and does not join.

Beware of a difference between versions. The old static scheme lists all the controllers in controller.quorum.voters (format id@host:port), which in KRaft version 1 is declared deprecated; with the dynamic quorum (KIP-853, active with kraft.version=1) you set controller.quorum.bootstrap.servers instead and add or remove controllers with kafka-metadata-quorum.sh add-controller and remove-controller. To know which mode you are in, run kafka-features.sh --bootstrap-controller localhost:9093 describe: if kraft.version is missing or is 0 the quorum is static.

# Template to adapt (not run): the cluster ID is generated only once
CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

# First controller, on its own; the others are added dynamically afterwards
bin/kafka-storage.sh format --cluster-id "$CLUSTER_ID" --standalone \
  --config config/controller.properties

# New broker in an existing cluster: no extra flag
bin/kafka-storage.sh format --cluster-id "$CLUSTER_ID" \
  --config config/server.properties

# New controller in an existing cluster
bin/kafka-storage.sh format --cluster-id "$CLUSTER_ID" \
  --config config/server.properties --no-initial-controllers

The --standalone flag creates a first controller that is the only voter; --initial-controllers instead lets you start with several voters, but the value must be identical on all of them. A new broker is formatted with no additional flags, a new controller with --no-initial-controllers. The typical mistake is formatting a node with a cluster ID different from the cluster's: the node starts, but it is not in the same cluster.

Migrating from ZooKeeper without losing the cluster

The official path goes through 3.9, the last bridge release: the 4.4 KRaft page says that to migrate "you need to use a bridge release. The last bridge release is Kafka 3.9". The instructions are in the 3.9 documentation, with the brokers' version brought to 3.9.1 and inter.broker.protocol.version set to 3.9. The mechanism has five states, in the following order:

1. Initial phase: all brokers in ZooKeeper mode, ZooKeeper-based controller.
2. Initial load: a newly created KRaft quorum reads the metadata from ZooKeeper.
3. Hybrid phase: some brokers are still ZooKeeper, but the controller is KRaft.
4. Dual-write: all brokers are KRaft, but the controller keeps writing to ZooKeeper too.
5. Finalization: the metadata is no longer written to ZooKeeper.

The order is not interchangeable and not always reversible. After finalization you cannot go back to ZooKeeper, the documentation says so, and during the migration you must not change the metadata version. Dual-write exists to give you a margin: until you finalize, the old system still has an up-to-date copy. Do not finalize until you have verified that the cluster behaves as expected.

# Migration from ZooKeeper to KRaft (template to adapt, from the Kafka 3.9 documentation)
# 1. KRaft controller: same cluster ID as the existing cluster
process.roles=controller
node.id=3000
controller.quorum.bootstrap.servers=localhost:9093
controller.listener.names=CONTROLLER
listeners=CONTROLLER://:9093
zookeeper.metadata.migration.enable=true
zookeeper.connect=localhost:2181
inter.broker.listener.name=PLAINTEXT

# 2. Brokers still in ZooKeeper mode, restarted one at a time
broker.id=0
inter.broker.protocol.version=3.9
zookeeper.metadata.migration.enable=true
zookeeper.connect=localhost:2181
controller.quorum.bootstrap.servers=localhost:9093
controller.listener.names=CONTROLLER

# 3. Broker converted to KRaft: broker.id becomes node.id, ZooKeeper properties removed
process.roles=broker
node.id=0
controller.quorum.bootstrap.servers=localhost:9093
controller.listener.names=CONTROLLER

The lines that count in the first block are zookeeper.metadata.migration.enable=true, which puts the controller in migration mode, and zookeeper.connect, which lets it read from ZooKeeper; the controller must be formatted with the same cluster ID as the cluster (you read it from meta.properties or from ZooKeeper). In the second block the brokers are restarted one at a time with the same two properties plus the quorum address: when the last one restarts, the migration starts by itself and the controller logs "Completed migration of metadata from Zookeeper to KRaft". In the third block each broker is converted, broker.id becomes node.id, process.roles=broker is added and the ZooKeeper properties disappear; the ID must stay the same, and the controllers' node.id values must not coincide with any existing broker.id. Finally you finalize by removing the migration property from the controllers and restarting them one at a time. If you use authorization with AclAuthorizer, in the brokers' step it must be replaced with StandardAuthorizer.

What to watch to know the quorum is healthy

The tool is kafka-metadata-quorum.sh describe --status. In the documentation the output shows fields such as LeaderId, LeaderEpoch, HighWatermark, MaxFollowerLag and the list of current voters. I do not repeat an example with invented values: look at its shape on the page and compare it with your cluster. A MaxFollowerLag that grows indicates a controller that is not keeping pace; a missing LeaderId indicates that no controller has been elected, typically because the majority is unreachable (network, listener, cluster ID).

To inspect the metadata there are two official tools: kafka-dump-log.sh --cluster-metadata-decoder decodes the segments and snapshots of __cluster_metadata-0, and kafka-metadata-shell.sh opens a snapshot like a file system, with directories such as brokers, topics and metadataQuorum. Beware: the file 00000000000000000000-0000000000.checkpoint does not contain cluster metadata and must not be used with the shell.

The configuration differences, in a table

The table is a map for whoever converts a server.properties file. The last row deserves a note: in KRaft the metadata version is raised with kafka-features.sh upgrade, only after all nodes have the new software, and it is not changed during a migration. The same documentation shows how to raise the features with upgrade --release-version 4.1 or, for the quorum alone, upgrade --feature kraft.version=1.

What you can do now

You can explain that in KRaft the metadata is a log replicated with Raft and that the brokers read it in pieces. You can choose the quorum size (3 or 5, never an even number) and prefer dedicated controllers in production. You can format a node with a cluster ID decided by you, and tell a static quorum from a dynamic one with a command. You know that to migrate from ZooKeeper you must go through 3.9 and that after finalization there is no going back. The next step is reliable writing towards those brokers, in the article on advanced producers and consumers; when you need to size a real cluster, go back to the article on sizing and disaster recovery. If you arrive from another point of the series, the links are: monitoring.