ZooKeeper vs KRaft in Apache Kafka

# ZooKeeper mode (Kafka 3.x and older)
broker.id=4
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181

# KRaft mode: dedicated controller
process.roles=controller
node.id=1
controller.quorum.voters=1@c1:9093,2@c2:9093,3@c3:9093
listeners=CONTROLLER://c1:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT

# KRaft mode: broker
process.roles=broker
node.id=4
controller.quorum.voters=1@c1:9093,2@c2:9093,3@c3:9093
listeners=PLAINTEXT://b4:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

public class ControllerFailoverModel {

    // Metadata changes, in the form KRaft appends them to __cluster_metadata
    sealed interface MetadataRecord permits BrokerRegistered, TopicCreated, LeaderChanged {}
    record BrokerRegistered(int brokerId) implements MetadataRecord {}
    record TopicCreated(String topic, int partitions) implements MetadataRecord {}
    record LeaderChanged(String topic, int partition, int newLeader) implements MetadataRecord {}

    // In-memory metadata that a controller builds by replaying the log
    static final class MetadataImage {
        final List<Integer> brokers = new ArrayList<>();
        final Map<String, Integer> leaders = new TreeMap<>(); // "topic-partition" -> leader
        int appliedRecords = 0;

        void apply(MetadataRecord rec) {
            switch (rec) {
                case BrokerRegistered b -> brokers.add(b.brokerId());
                case TopicCreated t -> {
                    for (int p = 0; p < t.partitions(); p++) {
                        leaders.put(t.topic() + "-" + p, brokers.get(p % brokers.size()));
                    }
                }
                case LeaderChanged c -> leaders.put(c.topic() + "-" + c.partition(), c.newLeader());
            }
            appliedRecords++;
        }
    }

    // ZooKeeper mode: a new controller reads the broker list, every broker znode,
    // the topic list, every topic znode and every partition state znode
    static long zooKeeperReads(int brokers, int topics, long partitions) {
        return 1 + brokers + 1 + topics + partitions;
    }

    public static void main(String[] args) {
        List<MetadataRecord> log = List.of(
                new BrokerRegistered(101), new BrokerRegistered(102), new BrokerRegistered(103),
                new TopicCreated("orders", 6), new TopicCreated("payments", 4),
                new TopicCreated("audit", 2),
                new LeaderChanged("orders", 0, 102), new LeaderChanged("payments", 3, 103));

        // A standby controller follows the leader; here it is 2 records behind
        MetadataImage standby = new MetadataImage();
        log.subList(0, log.size() - 2).forEach(standby::apply);
        System.out.println("Standby has applied " + standby.appliedRecords + " of " + log.size() + " records");

        // Active controller fails: the standby only replays the missing tail
        List<MetadataRecord> tail = log.subList(standby.appliedRecords, log.size());
        tail.forEach(standby::apply);
        System.out.println("KRaft failover: replayed " + tail.size() + " records -> "
                + standby.leaders.size() + " partitions ready");
        System.out.println("  orders-0 leader = " + standby.leaders.get("orders-0")
                + ", payments-3 leader = " + standby.leaders.get("payments-3"));

        System.out.println("ZooKeeper failover: " + zooKeeperReads(3, 3, 12)
                + " znode reads before the new controller can act");

        // Scale up: ZooKeeper work grows with partitions, KRaft work does not
        System.out.println("At 200,000 partitions: ZooKeeper reads = "
                + zooKeeperReads(30, 2_000, 200_000) + ", KRaft replay = " + tail.size() + " records");
    }
}

Standby has applied 6 of 8 records
KRaft failover: replayed 2 records -> 12 partitions ready
  orders-0 leader = 102, payments-3 leader = 103
ZooKeeper failover: 20 znode reads before the new controller can act
At 200,000 partitions: ZooKeeper reads = 202032, KRaft replay = 2 records

import java.util.Map;
import java.util.stream.Collectors;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.common.Node;

public class ControllerIdCheck {

    public static void main(String[] args) throws Exception {
        // Normal clients talk to brokers
        Map<String, Object> viaBrokers = Map.of(
                AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "b4:9092,b5:9092");
        // Admin clients can also talk to the controllers directly (Kafka 3.7+)
        Map<String, Object> viaControllers = Map.of(
                AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG, "c1:9093,c2:9093,c3:9093");

        try (Admin brokers = Admin.create(viaBrokers);
             Admin controllers = Admin.create(viaControllers)) {

            int raftLeader = brokers.describeMetadataQuorum().quorumInfo().get().leaderId();
            System.out.println("Raft leader (real active controller): " + raftLeader);

            print("bootstrap.servers    ", brokers.describeCluster());
            print("bootstrap.controllers", controllers.describeCluster());
        }
    }

    private static void print(String label, DescribeClusterResult cluster) throws Exception {
        Node controller = cluster.controller().get();
        String nodes = cluster.nodes().get().stream()
                .map(Node::id).sorted().map(String::valueOf)
                .collect(Collectors.joining(",", "[", "]"));
        System.out.println(label + " -> controller()=" + controller.id() + ", nodes()=" + nodes);
    }
}

Raft leader (real active controller): 2
bootstrap.servers     -> controller()=5, nodes()=[4,5]
bootstrap.controllers -> controller()=2, nodes()=[1,2,3]

import java.util.Map;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.QuorumInfo;

public class FailoverWatch {

    public static void main(String[] args) throws Exception {
        Map<String, Object> config = Map.of(
                AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "b4:9092,b5:9092",
                AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 2000,
                AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 2000);

        try (Admin admin = Admin.create(config)) {
            long start = System.currentTimeMillis();
            int lastLeader = -2;

            // Poll the quorum for 20 seconds and print every leader change
            while (System.currentTimeMillis() - start < 20_000) {
                long t = System.currentTimeMillis() - start;
                try {
                    QuorumInfo q = admin.describeMetadataQuorum().quorumInfo().get();
                    if (q.leaderId() != lastLeader) {
                        System.out.printf("t=%5d ms  leader=%d  epoch=%d  highWatermark=%d%n",
                                t, q.leaderId(), q.leaderEpoch(), q.highWatermark());
                        lastLeader = q.leaderId();
                    }
                } catch (Exception e) {
                    System.out.printf("t=%5d ms  no answer: %s%n", t, e.getCause().getClass().getSimpleName());
                    lastLeader = -1;
                }
                Thread.sleep(200);
            }
        }
    }
}

t=    0 ms  leader=2  epoch=3  highWatermark=248
t= 6958 ms  no answer: TimeoutException
t= 9180 ms  leader=3  epoch=5  highWatermark=262