KRaft контроллер: консенсус без ZooKeeper

Kafka 4.0 управляет метаданными кластера через KRaft (Kafka Raft) — встроенный консенсусный протокол на основе Raft. KRaft — не внешний сервис, а часть самого Kafka.

Почему убрали ZooKeeper

ZooKeeper (Apache ZooKeeper) выполнял роль распределённого хранилища метаданных: список брокеров, назначение лидеров партиций, конфигурация топиков. Это создавало несколько проблем:

Внешняя зависимость. Отдельный кластер ZooKeeper нужно было развёртывать, мониторить и обслуживать независимо от Kafka. Два отдельных кластера вместо одного.

Ограничение масштаба. ZooKeeper плохо масштабировался при большом числе партиций. Кластер с 200 000+ партиций имел заметные задержки leader election и controller failover (от 30 секунд до нескольких минут).

Двойная модель согласованности. Kafka-брокеры и ZooKeeper имели разные модели транзакционности — это усложняло обработку граничных случаев при сбоях.

KRaft устраняет все эти проблемы: метаданные хранятся внутри самих Kafka-узлов, масштабирование до 1 000 000+ партиций, failover менее чем за 20 секунд (vs минуты с ZooKeeper).

Raft consensus: кворум большинства

KRaft реализует протокол Raft. Ключевая идея: кворум большинства (majority quorum).

Для N-узлового quorum необходимо (N/2)+1 узлов для принятия решения (округление вверх). Примеры:

Стандарт для production: 3 или 5 voter-узлов. Чётное число (2, 4, 6) не рекомендуется: при split-brain кластер 4/2 не может принять решение (ни одна сторона не имеет большинства из 3), тогда как при 2/2 ни одна сторона не имеет большинства из 3.

Metadata log: event-sourced журнал состояния кластера

Metadata log — это append-only commit log (та же структура, что и пользовательские топики Kafka), в котором хранится вся история изменений состояния кластера.

Каждое изменение — новая запись в metadata log:

offset 0: RegisterBroker { nodeId: 1, host: "broker1", ... }
offset 1: RegisterBroker { nodeId: 2, host: "broker2", ... }
offset 2: CreateTopic { name: "orders", partitions: 6, rf: 3 }
offset 3: PartitionRecord { topic: "orders", partition: 0, leader: 1, isr: [1,2,3] }
...
offset 42: LeaderChange { topic: "orders", partition: 0, newLeader: 2 }

Это не реляционная таблица. Текущее состояние восстанавливается воспроизведением (replay) всего лога от начала. Для ускорения создаются снапшоты (snapshots) — сохранённое состояние на определённый offset. При запуске брокер загружает последний снапшот и воспроизводит только новые записи.

Инструмент для просмотра состояния KRaft quorum:

kafka-metadata-quorum.sh \
  --bootstrap-server localhost:9092 \
  describe --status

# Вывод:
# ClusterId:              abc123
# LeaderId:               1
# LeaderEpoch:            5
# HighWatermark:          1042
# MaxFollowerLag:         0
# MaxFollowerLagTimeMs:   12
# CurrentVoters:          [1,2,3]
# CurrentObservers:       []

KRaft leader election: последовательность событий

Когда контроллер-лидер недоступен (сбой, перегрузка, сетевая проблема), оставшиеся voter-узлы проводят новые выборы. Посмотрим на протокол выборов:

Epoch fencing: защита от split-brain

Epoch — монотонно возрастающий счётчик, присваиваемый каждому новому лидеру контроллера. При каждых новых выборах epoch увеличивается на единицу.

Проблема split-brain: старый лидер (не знающий, что его заменили) продолжает рассылать команды брокерам. Без защиты — два контроллера одновременно управляют кластером, что приводит к несогласованности метаданных.

Epoch fencing решает это просто: каждая metadata-команда включает epoch отправителя. Брокер отклоняет любую команду с epoch, меньшим чем текущий известный epoch:

Старый лидер (epoch 5): LeaderChange { epoch: 5, ... }
Брокер (знает epoch 6): "Epoch 5 < 6 — отклонено. Команда устаревшего лидера."

Новый лидер (epoch 6): LeaderChange { epoch: 6, ... }
Брокер (знает epoch 6): "Epoch 6 = 6 — принято."

Epoch fencing гарантирует: в любой момент времени команды принимаются только от текущего легитимного лидера. Старый лидер автоматически становится безвредным.

KIP-853: динамическое членство в quorum

KIP-853 (Kafka 4.0) добавляет возможность изменять состав KRaft voter-узлов без остановки кластера.

До KIP-853: изменение quorum (добавление или замена controller-узла) требовало остановки кластера. Это делало операции по масштабированию control plane болезненными.

После KIP-853: новый voter-узел добавляется через kafka-metadata-quorum.sh:

# Добавить узел node.id=4 в quorum (без рестарта)
kafka-metadata-quorum.sh \
  --bootstrap-server localhost:9092 \
  add-controller \
  --controller-id 4 \
  --controller-directory-id <directory-id>

Новый узел сначала выступает как observer (получает metadata log, но не голосует), затем переходит в voter статус после достижения кворума существующих voter-узлов. Это полностью онлайн-операция.

Конфигурация KRaft в production

Минимальная конфигурация для dedicated controller-узла (Kafka 4.0):

# Controller-only node: не хранит пользовательские данные
process.roles=controller
node.id=1

# KRaft quorum: 3 voter-узла
controller.quorum.voters=1@controller1:9093,2@controller2:9093,3@controller3:9093

# Controller listener (не для клиентов)
listeners=CONTROLLER://0.0.0.0:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT

# Metadata log storage
log.dirs=/var/kafka/metadata

Ключевые выводы

1. KRaft — встроенный Raft-протокол, заменивший ZooKeeper в Kafka 4.0. zookeeper.connect не существует.
2. Metadata log — append-only event-sourced лог всех изменений кластера (не реляционная таблица).
3. Raft quorum: нечётное число voter-узлов (3 или 5). Кворум = большинство.
4. Epoch монотонно возрастает при каждых выборах. Epoch fencing отклоняет команды от устаревшего лидера.
5. KIP-853: добавление/удаление controller-узлов без остановки кластера (Kafka 4.0+).