KRaft and the controller

What it is and why it exists

A Kafka cluster has metadata: the list of topics, the replicas and leader of every partition, every setting, and the list of brokers. All brokers must agree on it. If two brokers both believed they led the same partition, data would be lost.

Before Kafka 4, a separate system named ZooKeeper held the metadata. Kafka 4.0 removed it. Kafka now stores its metadata in itself, with a design named KRaft: Kafka plus Raft. Raft is a well-known method for several servers to agree on one ordered log, as long as more than half of them can talk to each other.

The servers that run it are controllers. They form the quorum, the voting group. One of them is elected leader and is the active controller. It makes every metadata decision and writes it as a record in the metadata log. The other controllers copy that log and are ready to take over. Brokers copy it too, and act on what they read.

Failure scenarios

Hands-on lab

k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
B=kafka-1:19092

Step 1. Look at the quorum

k kafka-metadata-quorum.sh --bootstrap-server $B describe --status
k kafka-metadata-quorum.sh --bootstrap-server $B describe --replication

ClusterId:              4L6g3nShT-eMCtK--X86sw
LeaderId:               1
LeaderEpoch:            3
HighWatermark:          5717
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   0
CurrentVoters:          [{"id": 1, …"CONTROLLER://kafka-1:9093"}, {"id": 2, …}, {"id": 3, …}]
CurrentObservers:       []

NodeId  LogEndOffset  Lag  Status
1       5719          0    Leader
2       5719          0    Follower
3       5719          0    Follower

Node 1 is the active controller. The metadata log has about 5,700 records, and both followers have all of them. HighWatermark means the same as in module 04, with "majority" in place of "in-sync set".

Step 2. Read the identity files

docker exec kafka-1 cat /tmp/kraft-combined-logs/meta.properties
docker exec kafka-1 cat /tmp/kraft-combined-logs/__cluster_metadata-0/quorum-state

cluster.id=4L6g3nShT-eMCtK--X86sw
directory.id=3uiEyBOPVNJZjvqFZCMIdA
node.id=1
version=1

{"clusterId":"","leaderId":1,"leaderEpoch":3,"votedId":1,
 "currentVoters":[{"voterId":1},{"voterId":2},{"voterId":3}],"data_version":0}

The second file is this node's memory of the election: the current epoch, the leader, and whom it voted for. It is written to disk so that a restart cannot make the node vote twice in one epoch.

Step 3. Read the metadata log

The flag is --cluster-metadata-decoder. Count the record types, then look at single records.

dump() { docker exec kafka-2 /opt/kafka/bin/kafka-dump-log.sh --cluster-metadata-decoder \
  --files /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000.log; }
dump | grep -o '"type":"[A-Z_]*"' | sort | uniq -c | sort -rn | head
dump | grep -m1 'TOPIC_RECORD' | grep -o 'payload: .*'
dump | grep -m1 '"PARTITION_RECORD"' | grep -o 'payload: .*'
dump | grep -m1 'PARTITION_CHANGE_RECORD' | grep -o 'payload: .*'

4731 "type":"NO_OP_RECORD"
1533 "type":"PARTITION_CHANGE_RECORD"
  79 "type":"PARTITION_RECORD"
  51 "type":"BROKER_REGISTRATION_CHANGE_RECORD"
  23 "type":"CONFIG_RECORD"
  19 "type":"TOPIC_RECORD"

payload: {"type":"TOPIC_RECORD","version":0,"data":{"name":"orders","topicId":"mocIQ-cjTlOVeEC3U3NxRQ"}}

payload: {"type":"PARTITION_RECORD","version":2,"data":{"partitionId":0,"topicId":"mocIQ-cjTlOVeEC3U3NxRQ",
  "replicas":[2,3,1],"isr":[2,3,1],"removingReplicas":[],"addingReplicas":[],"leader":2,"leaderEpoch":0,
  "partitionEpoch":0,"directories":[…]}}

payload: {"type":"PARTITION_CHANGE_RECORD","version":2,"data":{"partitionId":0,"topicId":"dEFhkuV0…","isr":[2,1]}}

This is the orders topic from module 01, as the controller recorded it: first the topic, then each partition with its replicas and leader. The last record is an in-sync set that shrank to 2 and 1 when broker 3 was stopped in an earlier lab. A change record holds only the fields that changed.

Step 4. Stop the active controller

Node 1 is the leader, so run the tools from node 2.

docker stop kafka-1
docker exec -i kafka-2 /opt/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server kafka-2:19092 describe --status
docker exec -i kafka-2 /opt/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server kafka-2:19092 describe --replication
docker exec -i kafka-2 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-2:19092 \
  --create --topic during-failover --partitions 1 --replication-factor 2

LeaderId: 2   LeaderEpoch: 4            ← about 1 second after the stop

NodeId  LogEndOffset  Lag   Status
2       5875          0     Leader
1       -1            5876  Follower
3       5875          0     Follower

Created topic during-failover.

Node 2 is the new active controller, in a new epoch. Two of three controllers run, which is a majority, so the cluster accepted a new topic.

Step 5. Lose the majority

docker stop kafka-3
docker exec -i kafka-2 timeout 12 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-2:19092 \
  --describe --topic during-failover
docker exec -i kafka-2 timeout 14 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-2:19092 \
  --create --topic no-quorum --partitions 1 --replication-factor 1
docker logs --since 20s kafka-2 2>&1 | grep -i 'election' | tail -1

(no output: both commands ran into the time limit)

INFO [RaftManager id=2] Election timed out before receiving sufficient vote responses to become candidate. …

One controller of three cannot elect itself. It keeps asking for votes that nobody answers. Even reading a topic's description did not work, because the tool asks the cluster for current metadata.

Step 6. Restore

docker start kafka-1 kafka-3
# wait 20 seconds
k kafka-metadata-quorum.sh --bootstrap-server $B describe --status

LeaderId:        2
LeaderEpoch:     6
MaxFollowerLag:  0

Node 2 is leader again, now in epoch 6: the failed election attempts each used an epoch. The two returning nodes copied the records they had missed.

Step 7. See which brokers are registered

k kafka-broker-api-versions.sh --bootstrap-server $B | grep '^kafka-'
k kafka-features.sh --bootstrap-server $B describe

kafka-2:19092 (id: 2 rack: null isFenced: false) -> (
kafka-1:19092 (id: 1 rack: null isFenced: false) -> (
kafka-3:19092 (id: 3 rack: null isFenced: false) -> (

Feature: eligible.leader.replicas.version  SupportedMinVersion: 0  SupportedMaxVersion: 1  FinalizedVersionLevel: 1
Feature: metadata.version                  SupportedMinVersion: 3.3-IV3  SupportedMaxVersion: 4.3-IV0  …

Common mistakes and misconceptions

Staff-level questions

Self-test