Призрачные контроллеры Kafka в KRaft и как они блокируют апгрейд кластера

Призрачные контроллеры Kafka (Kafka ghost controllers) появляются, когда контроллер KRaft давно удалён и остановлен, а кластер продолжает считать его участником метаданных. Из-за этого может не пройти обновление фичи или metadata.version. Команда Aiven (управляемая платформа для open source систем, в том числе Kafka) опубликовала разбор такого случая и своё обходное решение. В статье мы разберём, чем кворум отличается от списка регистраций, соберём стенд из четырёх контроллеров и напишем скрипт, который находит расхождение между этими списками до апгрейда. Сразу предупредим, что прогон показал не то, что мы ожидали. Механизм из исходника, то есть сохранение регистрации после вывода узла из кворума, на нашем стенде воспроизвести не удалось, и ниже мы честно разбираем, что именно прогон показал.

Что произошло у Aiven

Исходник называется Exorcising Ghost Apache Kafka Controllers. Всё началось с обращения клиента. Свежеобновлённый кластер не включал новую фичу Share Group, а лидер кластера сопротивлялся изменению. В метаданных нашёлся контроллер, который давно удалили, но кластер всё ещё считал его своим. Состояние и поддерживаемые фичи этого узла, по рассказу авторов, были заморожены во времени, и именно они блокировали включение новой возможности.

По данным авторов, проблема появилась в Kafka 3.9, сильнее всего бьёт по версиям с 4.0 по 4.3, а штатное исправление доступно только в 4.4. Aiven утверждает, что под угрозой любой кластер, созданный или обновлённый за последние полтора года, и что для версии 4.0 официальных исправлений сообщества уже не будет. Сроки окончания поддержки мы самостоятельно не проверяли, поэтому дат не приводим. Все версии и сроки в этом разделе взяты из исходника, а не из документации Kafka.

Штатное решение сообщества, описанное в исходнике как KIP-1312 (Kafka Improvement Proposal, предложение по улучшению Kafka), добавляет команду unregister-controller. По словам Aiven, она дописывает в метаданные запись UnregisterControllerRecord и убирает регистрацию узла из образа метаданных. Номер и текст KIP мы берём только из исходника, саму страницу предложения мы не проверяли.

Дальше начинается самое интересное. По описанию авторов, команда закрыта гейтом по версии метаданных и работает только после перехода кластера на metadata.version 4.4-IV2. А старые регистрации призраков заявляют возможности прежней версии Kafka, и валидация фич учитывает их при переходе на 4.4. Получается замкнутый круг. Чтобы убрать призрака, нужно обновить версию метаданных, а чтобы её обновить, нужно сначала убрать призрака. Aiven сообщает, что проверила оба порядка и получила отказ в обоих.

Выход у них такой, и это пересказ их статьи. Патч для версий 4.0-4.3 использует существующую запись регистрации контроллера и добавляет маркерную фичу __decommissioned_controller, а пропатченные контроллеры пропускают помеченные регистрации при проверке фич и версии метаданных. В порте на 4.4, по словам авторов, добавлена и штатная запись снятия регистрации. Сами патчи мы не читали и их поведение не проверяли.

Voters и registrations в метаданных KRaft

Чтобы понять, откуда берётся призрак, нужно развести две сущности, которые легко принять за одну. Первая это кворум. По KIP-853 голосующий участник (voter) это реплика раздела метаданных, которая может стать кандидатом и лидером, а набор voters (set of voters) хранится в самом логе записью VotersRecord. Вывести контроллер из кворума можно командой kafka-metadata-quorum remove-controller, она отправляет лидеру запрос RemoveVoter. Лидер записывает новый набор voters, и после коммита контроллер перестаёт влиять на большинство.

Вторая сущность это регистрации (registrations). Каждый контроллер при старте сообщает кластеру свою версию и поддерживаемые фичи записью RegisterControllerRecord. Эти данные нужны, чтобы кластер мог проверить, что все участники умеют новую фичу или новую версию метаданных. Исходник утверждает, что остановленный и удалённый из voters контроллер сохраняет такую запись. В KIP-853 мы не нашли механизма, который снимал бы регистрацию вместе с удалением из кворума, но не нашли и явного описания обратного. Это утверждение исходника, и наш стенд его не подтвердил, потому что до вывода узла из кворума дело не дошло. Подробности в разделе практики.

Обе сущности живут в одном логе, но отвечают на разные вопросы. Кворум отвечает на вопрос, кто голосует. Регистрации отвечают на вопрос, кто и что умеет. Призрак это узел, который присутствует во втором списке и отсутствует в первом. Как именно он туда попал, для проверки расхождения не так важно, и стенд показал, что путей к такому состоянию больше одного.

Почему призрак мешает обновлению

По описанию Aiven, когда клиент или администратор просит поднять фичу, активный контроллер проверяет не только живых участников. Он сверяет запрошенный уровень с тем, что заявлено в регистрациях контроллеров. Если у призрака в записи указан старый максимальный уровень, проверка не проходит, хотя этого узла давно нет. Именно так у Aiven не включилась Share Group. Отдельно подчеркнём, что этот механизм мы не проверяли. Исходник описывает конкретно Share Group и переход на 4.4-IV2.

Замкнутый круг в Kafka 4.4

По исходнику, в 4.4 ситуация парадоксальна. Инструмент для снятия регистрации существует, но гейт по версии метаданных требует уровня 4.4-IV2, а призрак не даёт до него дойти. Поэтому Aiven разделила два действия. Сначала узел помечается как выведенный из эксплуатации через существующую запись регистрации, потом пропатченная валидация перестаёт его учитывать, и только затем возможна настоящая отмена регистрации.

Как справиться с призраком

У инженера немного вариантов, и все они упираются в версию Kafka. Таблица ниже собрана только из того, что сказано в исходнике.

Разбор простой. Если вы на 4.4 и в метаданных нет призраков старой версии, штатная команда выигрывает по всем статьям. Если призраки есть и вы на 4.0-4.3, остаётся форк с патчем, и это решение для вендоров и тех, кто уже собирает Kafka сам. Для обычной эксплуатации самое рациональное это увидеть призрака до апгрейда. Ручная подмена фич годится разве что как аварийная мера на один раз, потому что каждый новый уровень фичи потребует повторения. Основы работы с кворумом и версиями метаданных разбираются на курсе Администрирование кластера Kafka.

Практика на стенде

Стенд состоит из четырёх контроллеров KRaft и одного брокера на образе apache/kafka:4.3.2, всего около 4 ГБ памяти. Наружу мы ничего не публикуем, поэтому и портов на 127.0.0.1 нет, все команды выполняются внутри контейнеров через docker compose exec. Контроллер 1 форматируется с флагом —standalone, остальные с —no-initial-controllers и должны войти в кворум сами благодаря controller.quorum.auto.join.enable, как описано в KIP-853.

Сразу оговорим пределы стенда. Все контроллеры здесь одной версии 4.3.2, а образа с 4.4 в кэше нет. Поэтому мы могли воспроизвести только расхождение между voters и registrations. Реальную блокировку апгрейда и команду unregister-controller стенд не проверяет, для этого понадобился бы призрак более старой версии.

Начнём с файла compose.yaml.

# compose.yaml · Docker Compose · apache/kafka:4.3.2
x-controller: &controller
  image: apache/kafka:4.3.2
  mem_limit: 768m
  command:
    - bash
    - -c
    - |
      set -e
      cat > /tmp/controller.properties <<EOF
      process.roles=controller
      node.id=$${NODE_ID}
      controller.quorum.bootstrap.servers=controller1:9093
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT
      listeners=CONTROLLER://controller$${NODE_ID}:9093
      controller.quorum.auto.join.enable=true
      log.dirs=/tmp/kraft-logs
      EOF
      if [ "$${NODE_ID}" = "1" ]; then MODE=--standalone; else MODE=--no-initial-controllers; fi
      /opt/kafka/bin/kafka-storage.sh format --ignore-formatted --cluster-id $${CLUSTER_ID} --config /tmp/controller.properties $${MODE}
      exec /opt/kafka/bin/kafka-server-start.sh /tmp/controller.properties

services:
  controller1:
    <<: *controller
    hostname: controller1
    environment:
      NODE_ID: "1"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qg
      KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m"
  controller2:
    <<: *controller
    hostname: controller2
    depends_on: [controller1]
    environment:
      NODE_ID: "2"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qg
      KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m"
  controller3:
    <<: *controller
    hostname: controller3
    depends_on: [controller1]
    environment:
      NODE_ID: "3"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qg
      KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m"
  controller4:
    <<: *controller
    hostname: controller4
    depends_on: [controller1]
    environment:
      NODE_ID: "4"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qg
      KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m"
  broker:
    image: apache/kafka:4.3.2
    hostname: broker
    mem_limit: 1024m
    depends_on: [controller1]
    environment:
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qg
      KAFKA_HEAP_OPTS: "-Xms512m -Xmx512m"
    command:
      - bash
      - -c
      - |
        set -e
        cat > /tmp/broker.properties <<EOF
        process.roles=broker
        node.id=5
        controller.quorum.bootstrap.servers=controller1:9093,controller2:9093,controller3:9093
        controller.listener.names=CONTROLLER
        listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
        listeners=PLAINTEXT://broker:9092
        advertised.listeners=PLAINTEXT://broker:9092
        offsets.topic.replication.factor=1
        log.dirs=/tmp/kraft-logs
        EOF
        /opt/kafka/bin/kafka-storage.sh format --ignore-formatted --cluster-id $${CLUSTER_ID} --config /tmp/broker.properties
        exec /opt/kafka/bin/kafka-server-start.sh /tmp/broker.properties

Общую часть мы вынесли в якорь, поэтому каждый контроллер отличается только номером. Брокеру в качестве точек входа мы даём контроллеры 1-3, потому что контроллер 4 предполагалось выводить из кворума. Теперь файл requirements.txt. Сторонних пакетов не нужно, скрипты используют только стандартную библиотеку.

# requirements.txt

Общие функции лежат в ghostlib.py. Модуль читает кворум через kafka-metadata-quorum describe —status, а регистрации достаёт из лога метаданных утилитой kafka-dump-log с декодером метаданных. Другого способа прочитать регистрации контроллеров из командной строки мы не использовали. Формат вывода дампа зависит от версии, поэтому при пустом результате модуль не гадает, а останавливается с сырым выводом. Обратите внимание, что кворум читается через брокер, а регистрации из лога на контроллере 1. Это два независимых источника, и это важно для задания 2.

# ghostlib.py · Python 3 stdlib · Kafka 4.3.2
import os
import re
import subprocess

HERE = os.path.dirname(os.path.abspath(__file__))
BIN = "/opt/kafka/bin"
BOOTSTRAP = "broker:9092"
META_DIR = "/tmp/kraft-logs/__cluster_metadata-0"


def compose(*args):
    r = subprocess.run(["docker", "compose", *args], cwd=HERE, capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError("docker compose " + " ".join(args) + "\n" + r.stdout + r.stderr)
    return r.stdout


def run_in(service, *cmd):
    return compose("exec", "-T", service, *cmd)


def voters():
    out = run_in("broker", BIN + "/kafka-metadata-quorum.sh", "--bootstrap-server", BOOTSTRAP, "describe", "--status")
    for line in out.splitlines():
        if line.startswith("CurrentVoters:"):
            found = re.findall(r'"id"\s*:\s*(\d+)\s*,\s*"directoryId"\s*:\s*"([^"]+)"', line)
            if found:
                return {int(i): d for i, d in found}
    raise RuntimeError("CurrentVoters не разобран, сырой вывод:\n" + out)


def registrations():
    files = run_in("controller1", "bash", "-c", "ls " + META_DIR + "/*.log").split()
    out = run_in("controller1", BIN + "/kafka-dump-log.sh", "--cluster-metadata-decoder", "--files", ",".join(files))
    ids = set()
    for line in out.splitlines():
        if re.search(r"(?<![A-Z_])REGISTER_CONTROLLER_RECORD", line):
            m = re.search(r'"controllerId"\s*:\s*(\d+)', line)
            if m:
                ids.add(int(m.group(1)))
    if not ids:
        raise RuntimeError("записи регистрации контроллеров не найдены, проверьте формат вывода kafka-dump-log")
    return ids


def metadata_version():
    out = run_in("broker", BIN + "/kafka-features.sh", "--bootstrap-server", BOOTSTRAP, "describe")
    return [l.strip() for l in out.splitlines() if "metadata.version" in l]


def report(title):
    vs = voters()
    regs = registrations()
    ghosts = sorted(regs - set(vs))
    print("== " + title + " ==")
    print("voters:       ", sorted(vs))
    print("registrations:", sorted(regs))
    print("ghosts:       ", ghosts or "нет")
    for line in metadata_version():
        print(line)
    return ghosts

Основной сценарий в ghost_check.py до 120 секунд ждёт полный кворум из контроллеров 1-4. Если кворум не собрался, скрипт печатает сообщение и всё равно идёт дальше, печатает отчёт и только тогда, когда узел 4 числится в voters, выводит его через remove-controller и останавливает контейнер. После этого он печатает второй отчёт. Формулировка сообщения «считаем узел 4 уже выведенным» заложена в код заранее, и как мы увидим, на этом прогоне она вводит в заблуждение.

# ghost_check.py · Python 3 stdlib · Kafka 4.3.2
import time

from ghostlib import BIN, BOOTSTRAP, compose, metadata_version, report, run_in, voters

TARGET = 4
EXPECTED = {1, 2, 3, 4}


def wait_cluster(timeout=120):
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            if EXPECTED <= set(voters()) and metadata_version():
                return True
        except RuntimeError:
            pass
        time.sleep(3)
    return False


def main():
    if not wait_cluster():
        print("полный кворум 1-4 не собран за 120 с, считаем узел 4 уже выведенным")
    report("до вывода контроллера 4")
    current = voters()
    if TARGET in current:
        run_in("broker", BIN + "/kafka-metadata-quorum.sh", "--bootstrap-server", BOOTSTRAP,
               "remove-controller", "--controller-id", str(TARGET),
               "--controller-directory-id", current[TARGET])
        compose("stop", "controller" + str(TARGET))
        time.sleep(5)
        print("контроллер " + str(TARGET) + " удалён из voters и остановлен")
    report("после вывода контроллера 4")


main()

Запускает всё это run.sh. Он поднимает стенд, создаёт окружение и запускает основной пример, а гасить стенд будет раннер.

# run.sh · Docker Compose · Python 3 · Kafka 4.3.2
set -euo pipefail
cd "$(dirname "$0")"
docker compose up -d
python3 -m venv .venv
.venv/bin/python -m pip install --quiet -r requirements.txt
.venv/bin/python ghost_check.py

Вывод основного примера ниже.

полный кворум 1-4 не собран за 120 с, считаем узел 4 уже выведенным
== до вывода контроллера 4 ==
voters:        [1, 2, 3]
registrations: [1, 2, 3, 4]
ghosts:        [4]
Feature: metadata.version                          SupportedMinVersion: 3.3-IV3          SupportedMaxVersion: 4.3-IV0          FinalizedVersionLevel: 4.3-IV0          Epoch: 707
== после вывода контроллера 4 ==
voters:        [1, 2, 3]
registrations: [1, 2, 3, 4]
ghosts:        [4]
Feature: metadata.version                          SupportedMinVersion: 3.3-IV3          SupportedMaxVersion: 4.3-IV0          FinalizedVersionLevel: 4.3-IV0          Epoch: 726

Прогон дал не тот результат, на который рассчитывал сценарий. Полный кворум из четырёх контроллеров за 120 секунд не собрался. Контроллер 4 успел зарегистрироваться, поэтому он есть в registrations, но в набор voters не вошёл, и в обоих отчётах voters равен [1, 2, 3]. Условие вывода узла из кворума не сработало, значит remove-controller и остановка контейнера не выполнялись. Оба отчёта различаются только значением Epoch в строке metadata.version, а сообщение о том, что узел уже выведен, на самом деле означает «в voters его никогда не было».

Что из этого следует. Строка ghosts действительно показывает расхождение между кворумом и регистрациями, и скрипт его находит. Но это расхождение получилось не так, как в исходнике, где узел сначала был voter, а потом покинул кворум. Сохранение регистрации после remove-controller наш стенд не воспроизвёл и не доказал. Почему контроллер 4 не вошёл в кворум автоматически, мы не выяснили. Причина может быть во времени старта, в настройках автоприсоединения или в чём-то ещё, а проверить это можно, сравнив логи контроллера 4 с выводом kafka-metadata-quorum describe —status на более долгом ожидании. Строка с metadata.version остаётся на 4.3-IV0, потому что фичу мы не поднимаем и блокировку здесь не воспроизводим.

Задания

Все решения используют ghostlib.py из практики и запускаются на том же стенде. Помните, что состояние стенда у нас именно такое, как в прогоне выше, с узлом 4 в регистрациях и вне voters.

Задание 1. Скрипт для CI

Напишите скрипт, который печатает в одну строку идентификаторы контроллеров, зарегистрированных, но не входящих в voters. Скрипт должен уметь вернуть ненулевой код выхода, чтобы им можно было останавливать конвейер.

# ghosts_only.py · Python 3 stdlib · Kafka 4.3.2
import sys

from ghostlib import registrations, voters

ghosts = sorted(registrations() - set(voters()))
print(" ".join(str(g) for g in ghosts) or "none")
if "--strict" in sys.argv and ghosts:
    sys.exit(1)

4

Задание 2. Остановка без remove-controller

Остановите контроллер 3, не выводя его из кворума, и сравните отчёт с тем, что вы видели для контроллера 4. Объясните, чем недоступный voter должен отличаться от призрака.

# compare_stop.py · Python 3 stdlib · Kafka 4.3.2
import time

from ghostlib import compose, report

compose("stop", "controller3")
time.sleep(5)
report("контроллер 3 остановлен без remove-controller")

== контроллер 3 остановлен без remove-controller ==
voters:        [1, 2, 3]
registrations: [1, 2, 3, 4]
ghosts:        [4]
Feature: metadata.version                          SupportedMinVersion: 3.3-IV3          SupportedMaxVersion: 4.3-IV0          FinalizedVersionLevel: 4.3-IV0          Epoch: 763

Задание 3. Предпроверка перед апгрейдом

Оформите чек-лист перед повышением metadata.version как скрипт с тремя проверками. Призраков в регистрациях быть не должно, число voters должно быть нечётным и не меньше трёх, а версия метаданных должна читаться через kafka-features.

# preflight.py · Python 3 stdlib · Kafka 4.3.2
import sys

from ghostlib import metadata_version, registrations, voters

vs = voters()
ghosts = sorted(registrations() - set(vs))
mv = metadata_version()

checks = [
    ("нет зарегистрированных узлов вне voters", not ghosts, "призраки " + str(ghosts)),
    ("voters нечётное число и не меньше трёх", len(vs) >= 3 and len(vs) % 2 == 1, "voters " + str(sorted(vs))),
    ("metadata.version читается", bool(mv), "; ".join(mv) or "нет данных"),
]

failed = 0
for name, ok, details in checks:
    print(("PASS " if ok else "FAIL ") + name + " (" + details + ")")
    failed += 0 if ok else 1

if "--strict" in sys.argv and failed:
    sys.exit(1)

FAIL нет зарегистрированных узлов вне voters (призраки [4])
PASS voters нечётное число и не меньше трёх (voters [1, 2, 3])
PASS metadata.version читается (Feature: metadata.version                          SupportedMinVersion: 3.3-IV3          SupportedMaxVersion: 4.3-IV0          FinalizedVersionLevel: 4.3-IV0          Epoch: 778)

Если вам предстоит обновлять реальные кластеры KRaft и вы хотите уверенно работать с кворумом, версиями метаданных и процедурами замены контроллеров, загляните на курс Администрирование кластера Kafka.

Референсные ссылки

• Exorcising Ghost Apache Kafka Controllers, разбор проблемы и патча от Aiven
• KIP-853 KRaft Controller Membership Changes, динамический кворум, add-controller и remove-controller
• Документация Apache Kafka, справочник по утилитам и конфигурации