Topic Rebalancing: kafka-reassign-partitions

Вы добавили новый брокер в кластер. Он запустился, зарегистрировался в KRaft-метаданных. И… стоит пустым. Kafka не перераспределяет данные автоматически. В отличие от HDFS, где NameNode сам балансирует блоки, Kafka требует явного действия администратора.

Инструмент: kafka-reassign-partitions.sh. Три фазы: генерировать план, выполнить, проверить. И критически важный параметр, без которого reassignment убьёт production кластер: --throttle.

Зачем нужна перебалансировка

Три фазы: generate, execute, verify

Фаза 1: Generate

# Шаг 1: Создать файл с топиками для перераспределения
cat > /tmp/topics.json << 'EOF'
{
  "topics": [
    {"topic": "orders"},
    {"topic": "payments"},
    {"topic": "user-events"}
  ],
  "version": 1
}
EOF

# Шаг 2: Сгенерировать план
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file /tmp/topics.json \
  --broker-list "1,2,3,4" \
  --generate \
  2>&1 | tee /tmp/reassignment-plan.txt

Команда выведет два JSON-блока:

1. Current partition replica assignment — текущее расположение
2. Proposed partition reassignment configuration — предложенный план

# Сохранить только предложенный план (второй блок JSON)
# В реальности нужно вручную скопировать из вывода
cat /tmp/proposed-plan.json

ОБЯЗАТЕЛЬНО просмотрите предложенный план перед выполнением. Убедитесь:

• Новый брокер (4) присутствует в replica lists
• Предпочтительный лидер (первый элемент replicas[]) распределён равномерно
• Нет “горящих” партиций (критически важных), которые могут временно стать under-replicated

Фаза 2: Execute

kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file /tmp/proposed-plan.json \
  --execute \
  --throttle 52428800
# 52428800 bytes/s = 50 MB/s

Команда запускает фоновое копирование данных. Процесс:

1. Новая реплика добавляется как фолловер (Out-of-Sync Replica)
2. Данные копируются: лидер -> новый фолловер
3. Когда фолловер догнал лидера — он входит в ISR
4. Старая реплика удаляется из ISR и с диска

Во время выполнения: UnderReplicatedPartitions временно вырастет — это нормально. Каждая перемещаемая партиция временно имеет дополнительную реплику (over-replicated), затем лишняя удаляется.

Фаза 3: Verify

kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file /tmp/proposed-plan.json \
  --verify

Пример вывода:

Status of partition reassignment:
Reassignment of partition orders-0 is completed.
Reassignment of partition orders-1 is completed.
Reassignment of partition payments-0 is still in progress.
Reassignment of partition payments-1 is still in progress.

Продолжайте проверять до тех пор, пока все партиции не покажут “completed”.

Формат JSON плана переназначения

{
  "version": 1,
  "partitions": [
    {
      "topic": "orders",
      "partition": 0,
      "replicas": [2, 3, 4],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 1,
      "replicas": [3, 4, 1],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 2,
      "replicas": [4, 1, 2],
      "log_dirs": ["any", "any", "any"]
    }
  ]
}

Важно о порядке replicas:

• Первый элемент массива = preferred leader
• Порядок остальных элементов = порядок репликации
• Равномерное распределение preferred leaders по брокерам обеспечивает равномерную лидерную нагрузку

log_dirs:

• "any" — брокер сам выбирает дисковый путь (стандартный случай)
• Конкретный путь (например "/data/disk2") — для JBOD (Multiple disks per broker, Kafka 4.0+)

Throttle: критически важный параметр

Без throttle partition reassignment потребляет ALL доступную сетевую и дисковую пропускную способность. Результат: producers получают TimeoutException, consumers накапливают lag, latency взлетает.

Как работает throttle:

# Установить throttle при выполнении
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file /tmp/proposed-plan.json \
  --execute \
  --throttle 52428800   # 50 MB/s replication speed limit

Под капотом это устанавливает на затронутых брокерах:

• leader.replication.throttled.rate=52428800 — лидер ограничивает скорость отдачи данных фолловерам
• follower.replication.throttled.rate=52428800 — фолловер ограничивает скорость получения

Удалить throttle после завершения:

# ДЛЯ КАЖДОГО затронутого брокера (1, 2, 3, 4)
for BROKER_ID in 1 2 3 4; do
  kafka-configs.sh \
    --bootstrap-server localhost:9092 \
    --entity-type brokers \
    --entity-name $BROKER_ID \
    --alter \
    --delete-config leader.replication.throttled.rate,follower.replication.throttled.rate
done

Забытый throttle после завершения reassignment замедляет обычную репликацию. Всегда удаляйте его после verify.

Выбор значения throttle:

Практическое правило: throttle = 50% от доступной NIC bandwidth / число параллельных reassignment.

# 1 Gbps NIC = 125 MB/s
# Занято обычным трафиком: ~30 MB/s
# Доступно для reassignment: 125 - 30 = 95 MB/s
# Throttle = 50% от доступного = 47.5 MB/s ≈ 50 MB/s (52428800 bytes)

# Для 10 Gbps NIC с 200 MB/s обычного трафика:
# Доступно: 1250 - 200 = 1050 MB/s
# Throttle = 50% = 525 MB/s (550502400 bytes)

Оценка времени завершения:

duration_hours = data_to_move_GB / throttle_MB_s / 1024 × 3600

# Пример: 500 GB данных при 50 MB/s
# duration = 500 GB × 1024 MB/GB / 50 MB/s = 10240 s ≈ 2.8 часа

Playbook: добавление нового брокера

Preferred leader election:

kafka-leader-election.sh \
  --bootstrap-server localhost:9092 \
  --election-type preferred \
  --all-topic-partitions

После reassignment партиция может иметь лидера, который не является preferred leader (первый элемент в replicas[]). Это происходит если текущий лидер не изменился, но reassignment изменил порядок реплик. Preferred leader election перемещает лидерство к preferred реплике — без копирования данных, мгновенно.

Playbook: вывод брокера из эксплуатации

# Шаг 1: Сгенерировать план, исключив выводимый брокер
# broker-list не включает broker-3 (выводимый)
cat > /tmp/decommission-topics.json << 'EOF'
{"topics": [{"topic": "orders"}, {"topic": "payments"}], "version": 1}
EOF

kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file /tmp/decommission-topics.json \
  --broker-list "1,2,4" \  # broker-3 НЕ включён
  --generate

# Шаг 2: Execute с throttle
# Шаг 3: Verify до завершения
# Шаг 4: Убедиться что UnderReplicatedPartitions вернулся к 0
# Шаг 5: Остановить broker-3
bin/kafka-server-stop.sh

# Шаг 6: Удалить конфиг broker-3 из cluster (опционально, KRaft сам обнаружит отсутствие)

НИКОГДА не останавливайте брокер до переноса всех его партиций. Результат: UnderReplicatedPartitions взлетит, часть партиций может стать offline.

Мониторинг reassignment

Пока выполняется reassignment, смотрите эти метрики:

Kafka 4.0: JBOD и intra-broker reassignment

В Kafka 4.0 поддержка JBOD (Just a Bunch of Disks) значительно улучшена. Один брокер может иметь несколько disks:

# server.properties
log.dirs=/data/disk1,/data/disk2,/data/disk3

kafka-reassign-partitions.sh поддерживает перемещение партиций между дисками на одном брокере (intra-broker reassignment). Это быстрее межброкерского перемещения (нет сетевого трафика — только disk-to-disk copy):

{
  "version": 1,
  "partitions": [
    {
      "topic": "orders",
      "partition": 0,
      "replicas": [1, 2, 3],
      "log_dirs": ["/data/disk2", "any", "any"]
    }
  ]
}

Запись /data/disk2 для broker-1 означает: переместить партицию 0 топика orders на disk2 брокера 1. Полезно для балансировки нагрузки между дисками при неравномерном использовании.