Что отвечает за регистрацию Kafka между брокерами

Что отвечает за регистрацию Kafka между брокерами

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

Роль ZooKeeper в Kafka

ZooKeeper выполняет несколько важных функций:

1. Регистрация брокеров — каждый брокер регистрирует себя в ZooKeeper при запуске
2. Координация лидера — выбор лидера для каждого партишна
3. Управление конфигурацией — хранение конфигов топиков и брокеров
4. Координация потребителей — отслеживание групп потребителей и их смещений
5. Обнаружение сбоев — детектирование отказавших брокеров

Архитектура Kafka с ZooKeeper

    ┌─────────────────────────────────────────────────┐
    │           ZooKeeper Ensemble (3-5 нод)         │
    │  (хранит метаданные, координирует брокеры)     │
    └─────────┬──────────────────────────────────┬───┘
              │                                  │
    ┌─────────▼────────┐  ┌────────────────┐  ┌─▼──────────────┐
    │  Kafka Broker 1  │  │ Kafka Broker 2 │  │ Kafka Broker 3 │
    │  (partition 0,1) │  │  (partition 2) │  │  (partition 3) │
    └──────────────────┘  └────────────────┘  └────────────────┘
             │                    │                     │
    ┌────────▼────────────────────▼─────────────────────▼────────┐
    │                    Kafka Cluster                           │
    │  Topic: "orders" с 4 партишнами, replication factor = 3   │
    └──────────────────────────────────────────────────────────┘
             ▲                    │                     ▲
    ┌────────┴──────┐  ┌─────────▼───────┐  ┌────────┴──────┐
    │  Producer 1  │  │  Consumer Group  │  │  Producer 2  │
    │ (отправляет) │  │  (читает)        │  │ (отправляет) │
    └───────────────┘  └──────────────────┘  └───────────────┘

Процесс регистрации брокера в ZooKeeper

Когда Kafka брокер стартует, он:

// 1. Генерирует уникальный broker.id (обычно из конфига)
broker.id=1

// 2. Читает конфиг server.properties
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
listeners=PLAINTEXT://broker1:9092

// 3. Регистрирует себя в ZooKeeper в пути:
// /brokers/ids/1

// 4. Создаёт ephemeral node с информацией о себе:
{
  "jmx_port": 9999,
  "timestamp": "1635273600000",
  "endpoints": ["PLAINTEXT://broker1:9092"],
  "host": "broker1",
  "version": 4,
  "port": 9092
}

// 5. ZooKeeper отслеживает этот node
// Если брокер отключится, node удалится автоматически

Пути в ZooKeeper для Kafka

/brokers
  /ids              # Все активные брокеры
    /1              # Информация брокера с ID 1
    /2              # Информация брокера с ID 2
    /3              # Информация брокера с ID 3

/topics
  /orders           # Конфиг топика "orders"
    /partitions
      /0            # Партишн 0
      /1            # Партишн 1
      /2            # Партишн 2

/brokers/topics
  /orders           # Какие брокеры хранят реплики топика "orders"
    /1              # [0, 2, 1]  - партишн 0: лидер 1, реплики [0, 2]
    /2              # [2, 1, 0]  - партишн 1: лидер 2, реплики [1, 0]

/controller        # Контроллер кластера (координирует выборы лидеров)
  /epoch           # Версия контроллера

Выборы лидера партишна

Когда лидер партишна падает, ZooKeeper координирует выбор нового лидера:

// Партишн 0 топика "orders":
// Реплики: [1, 2, 3]  (брокер 1 - лидер)

// Брокер 1 падает...

// ZooKeeper замечает, что лидер не отвечает
// Переизбирает лидера из in-sync replicas (ISR)

// Новый порядок реплик: [2, 1, 3]  (брокер 2 - новый лидер)

ZooKeeper Ensemble (кластер)

Для надёжности ZooKeeper работает в режиме ensemble (обычно 3 или 5 нод):

# zookeeper-1.properties
server.1=zookeeper-1:2888:3888
server.2=zookeeper-2:2888:3888
server.3=zookeeper-3:2888:3888

# зookeeper-2.properties
server.1=zookeeper-1:2888:3888
server.2=zookeeper-2:2888:3888
server.3=zookeeper-3:2888:3888

Консенсус: минимум (N/2 + 1) нод должно быть живыми:

• 3 ноды: нужны 2 живые (может упасть 1)
• 5 нод: нужны 3 живые (может упасть 2)

Координация потребителей в ZooKeeper

Consumer group отслеживает смещения (offsets) в ZooKeeper:

// Путь смещения в ZooKeeper:
/consumers/my-group/offsets/my-topic/0  -> 12345

// Это означает: группа "my-group" прочитала до сообщения 12345 в партишне 0

// При перезапуске потребитель начнёт с сообщения 12346

Конфигурация Kafka для работы с ZooKeeper

# server.properties (Kafka брокер)

# Обязательно: подключение к ZooKeeper
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181

# Уникальный ID брокера в кластере
broker.id=1

# Хост и порт
listeners=PLAINTEXT://kafka1:9092
advertised.listeners=PLAINTEXT://kafka1:9092

# Сессия ZooKeeper
zookeeper.session.timeout.ms=18000

Команды для проверки регистрации

# Подключиться к ZooKeeper
zkCli.sh -server zk1:2181

# Список активных брокеров
ls /brokers/ids
# Результат: [1, 2, 3]

# Информация о брокере
get /brokers/ids/1
# Результат: {"jmx_port": 9999, "host": "kafka1", "port": 9092, ...}

# Список топиков
ls /topics
# Результат: [orders, payments, users, ...]

# Информация о топике
get /topics/orders
# Результат: конфиг топика

# Контроллер кластера
get /controller
# Результат: {"version": 1, "brokerid": 2, "timestamp": "..."}

Отслеживание статуса брокеров программно

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;

public class KafkaClusterMonitor {
    
    public static void main(String[] args) throws Exception {
        AdminClient admin = AdminClient.create(
            Collections.singletonMap("bootstrap.servers", "kafka1:9092")
        );
        
        // Получить информацию о кластере
        DescribeClusterResult result = admin.describeCluster();
        
        // Список всех брокеров
        result.brokers().get().forEach(broker -> {
            System.out.println("Broker ID: " + broker.id());
            System.out.println("Host: " + broker.host());
            System.out.println("Port: " + broker.port());
            System.out.println("Rack: " + broker.rack());
        });
        
        // Контроллер
        System.out.println("Controller: " + result.controller().get());
        
        admin.close();
    }
}

ZooKeeper vs Kraft (новый режим)

Apache Kafka 3.3+ может работать БЕЗ ZooKeeper, используя KRaft (Kafka Raft):

# KRaft режим вместо ZooKeeper
# В server.properties вместо zookeeper.connect:

process.roles=broker
node.id=1
controller.quorum.voters=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093

ZooKeeper vs KRaft:

Важные моменты

1. Ephemeral nodes — узлы, которые удаляются, когда клиент отключится:

Брокер 1 подключается -> создаёт /brokers/ids/1
Брокер 1 падает -> /brokers/ids/1 удаляется автоматически
Консьюмеры узнают, что брокер 1 мёртв

1. Quorum — большинство нод в ZooKeeper ensemble должны быть живы:

3 ноды: 2 живы - OK
3 ноды: 1 жива - NOT OK (нет quorum)

1. Координация между брокерами — через ZooKeeper:

Брокер 1: я стал лидером партишна 0
         -> пишет в ZooKeeper
Брокер 2: видит обновление в ZooKeeper
         -> знает, что тепер слать запросы брокеру 1

Вывод: ZooKeeper отвечает за регистрацию Kafka брокеров, выборы лидеров, координацию партишнов и потребителей. Это критическая компонента Kafka, хотя в новых версиях её функции переходят на встроенный KRaft контроллер.