Что такое rebalancing и когда он происходит

Ребалансировка решает две главные задачи:

🟢 Junior Level

30-секундный ответ

Rebalancing (ребалансировка) — это распределенный протокол координации, в ходе которого Kafka динамически переназначает партиции топика между участниками одной Consumer Group.

Ребалансировка решает две главные задачи:

Масштабируемость: распределить нагрузку поровну при добавлении новых инстансов сервиса.
Отказоустойчивость: автоматически забрать партиции у упавшего инстанса и передать их живым консьюмерам.

Главная цена классического ребаланса: кратковременная пауза в обработке данных (Stop-the-World), пока консьюмеры согласуют новое распределение.

Наглядная аналогия из жизни

Представьте бригаду курьеров службы доставки:

В смене работают 4 курьера, каждому назначен свой район города (партиция).
Если один курьер заболел и уехал домой, диспетчер (Group Coordinator) останавливает распределение заказов на 1 минуту, перекраивает карту и распределяет район заболевшего между оставшимися тремя курьерами.
Если на смену заступил пятый курьер, диспетчер забирает по одной улице у каждого и отдает новичку.

Когда происходит ребалансировка?

1. Изменение состава группы:
   ├── Добавлен новый консьюмер (scale up)
   ├── Консьюмер штатно завершил работу (SIGTERM, scale down)
   └── Консьюмер аварийно упал (OOM, сбой сети, kill -9)

2. Проблемы со здоровьем консьюмера:
   ├── Не пришел Heartbeat за session.timeout.ms (завис процесс, долгий GC)
   └── Время между вызовами poll() превысило max.poll.interval.ms

3. Изменение топологии топиков:
   ├── В топик добавлены новые партиции
   └── Консьюмер подписан по RegExp (например, "orders-.*") и появился новый подходящий топик

🟡 Middle Level

Эволюция протоколов: Eager vs Cooperative Rebalancing

В истории Kafka применяются два принципиально разных протокола:

Eager Rebalancing (устаревший, до Kafka 2.4):
- Механизм Stop-the-World.
- Все консьюмеры группы полностью бросают (revoke) свои текущие партиции и останавливают чтение.
- Выполняется глобальное перераспределение с нуля.
- Простой группы может достигать от 5 до 40 секунд!
Cooperative Sticky Rebalancing (рекомендуемый современный стандарт):
- Инкрементальное перераспределение.
- Консьюмеры отдают только те партиции, которые реально нужно переместить другому инстансу.
- Все остальные партиции продолжают непрерывно вычитываться без малейшей паузы.

# Включение кооперативного ребаланса в consumer.properties:
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

ConsumerRebalanceListener: фиксация позиции

При отборе партиции критически важно сбросить накопленный прогресс в брокер, иначе новый владелец партиции начнет читать старые данные:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.warn("У консьюмера отбираются партиции: {}. Выполняем экстренный коммит!", partitions);
        // Обязательно сохраняем обработанный прогресс перед потерей партиции
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Консьюмеру назначены новые партиции: {}. Инициализация локальных кэшей...", partitions);
    }
});

Две линии защиты здоровья: Heartbeat vs Poll Interval

Консьюмер отслеживается брокером по двум независимым таймерам:

                      Жизненный цикл мониторинга консьюмера
      ┌─────────────────────────────────────────────────────────────────┐
      │                                                                 │
      ▼                                                                 ▼
[ Фоновый Heartbeat поток ]                             [ Главный рабочий поток ]
- Отправляет heartbeat.interval.ms (3с)                 - Вызывает consumer.poll()
- Лимит: session.timeout.ms (45с)                       - Лимит: max.poll.interval.ms (300с)
- Проверяет: "Жива ли JVM и есть ли сеть?"              - Проверяет: "Не зависла ли бизнес-логика?"

🔴 Senior Level

Внутренний протокол координации группы

Ребалансировка управляется брокером — Group Coordinator (определяется как лидер партиции топика __consumer_offsets, куда хэшируется group.id):

Фаза FindCoordinator: Консьюмер узнает, какой именно брокер отвечает за его группу.
Фаза JoinGroup:
- Все консьюмеры отправляют запрос JoinGroupRequest.
- Координатор выбирает одного из них Лидером группы (обычно первый подключившийся).
- Координатор возвращает Лидеру список всех участников и поддерживаемые ими стратегии распределения.
Фаза назначения (Assignment):
- Лидер группы локально выполняет алгоритм распределения (PartitionAssignor).
- Брокер не считает распределение сам (в классическом протоколе), отдавая эту работу клиенту-лидеру!
Фаза SyncGroup:
- Лидер отправляет сформированный план распределения координатору (SyncGroupRequest).
- Остальные консьюмеры шлют пустой SyncGroupRequest и в ответе получают свои назначенные партиции.

Static Membership (`group.instance.id`): спасение Kubernetes

При стандартном деплое в K8s (Rolling Update):

Pod-1 перезагружается -> триггерится Rebalance.
Pod-1 поднялся -> триггерится Rebalance.
И так для всех 20 подов: кластер находится в непрерывной ребалансировке полчаса.

Решение (KIP-345 Static Membership):

# Присвоение консьюмеру уникального постоянного идентификатора инстанса
group.instance.id=order-service-pod-0

При рестарте пода координатор не запускает ребаланс сразу.
Он ждет в течение session.timeout.ms (например, 2 минуты).
Когда новый под с тем же group.instance.id подключается, координатор мгновенно возвращает ему его прежние партиции без ребаланса группы!

Архитектурная эволюция: KIP-848 (Next Generation Protocol)

В Apache Kafka 3.7+ и 4.0 внедряется революционный протокол Server-Side Rebalance:

Вычисление распределения переносится с Java-клиента на сторону брокера (Group Coordinator).
Устраняются фазы JoinGroup и SyncGroup.
Общение с брокером переведено на единый RPC-метод ConsumerGroupHeartbeat.
Ребалансировки становятся полностью прозрачными, фоновыми и на порядок быстрее.

4 Tricky Questions

1. Почему устаревший Eager Rebalancing вызывал паузы Stop-the-World для всей группы, и как Cooperative Rebalancing устраняет эту проблему?

Ответ:

В Eager протоколе ребаланс происходил в один раунд: при любом триггере абсолютно все консьюмеры сбрасывали владение всеми партициями (revoke). В течение нескольких секунд ни одна партиция в группе не читалась.
В Cooperative (KIP-429) протокол разбит на 2 раунда:
1. Раунд 1: Консьюмеры сообщают координатору свои текущие назначения. Координатор определяет, какие партиции нужно переместить. Только эти конкретные партиции отзываются у владельцев. Все остальные 90% партиций продолжают непрерывно обрабатываться!
2. Раунд 2: Отобранные партиции назначаются новым владельцам.
Итог: время простоя затрагивает только мигрирующие партиции и снижается с десятков секунд до сотен миллисекунд.

2. Почему фоновый поток успешно шлет Heartbeat, но брокер все равно выбрасывает консьюмера из группы с `CommitFailedException`?

Ответ:

Начиная с Kafka 0.10.1, отправка Heartbeat вынесена в отдельный фоновый daemon-поток.
Если фоновый поток жив, брокер знает, что JVM не упала (session.timeout.ms не сработал).
Однако жизненность обработки бизнес-логики контролируется в главном потоке параметром max.poll.interval.ms (дефолт 5 минут).
Если главный поток взял пачку из 500 записей и завис на обработке тяжелой транзакции на 6 минут, вызов consumer.poll() не происходит вовремя.
Координатор фиксирует факт зависания обработчика и инициирует ребаланс по таймауту опроса, исключая зависший инстанс из группы.

3. Что такое каскадный шторм ребалансировок (Rebalance Storm) и как разорвать этот замкнутый круг?

Ответ:

Анатомия катастрофы:
1. На сервис пришел всплеск нагрузки. Консьюмер $C_1$ берет слишком большой батч и не успевает уложиться в max.poll.interval.ms.
2. Координатор исключает $C_1$ и запускает ребаланс, передавая его партиции консьюмерам $C_2$ и $C_3$.
3. Теперь $C_2$ и $C_3$ получают в два раза больше нагрузки, захлебываются, превышают свой max.poll.interval.ms и тоже вылетают из группы!
4. Группа входит в бесконечный шторм ребалансов: обработка парализована на 100%.
Как разорвать:
- Уменьшить max.poll.records (например, со 1000 до 50).
- Увеличить max.poll.interval.ms.
- Включить CooperativeStickyAssignor.
- Добавить circuit breaker или таймаут на обработку одного сообщения.

4. Почему нельзя хранить изменяемое не сохраняемое состояние в памяти консьюмера без реализации `ConsumerRebalanceListener`?

Ответ:

Если консьюмер накапливает оконные агрегаты или локальный стейт (например, сессии пользователей в HashMap) и у него внезапно отзывают партицию в результате ребаланса:
1. Партиция передается другому узлу, у которого в памяти этого стейта нет.
2. Если старый узел не сохранил оффсеты и не сбросил стейт в постоянное хранилище в методе onPartitionsRevoked(), данные будут вычислены с ошибками или потеряны.
3. Поэтому при ребалансе обязательны: коммит оффсетов и очистка/сброс локального состояния. Для stateful-обработки правильнее использовать Kafka Streams с RocksDB State Stores.

🎯 Шпаргалка для интервью

Главное о ребалансировке

Что это: перераспределение партиций между инстансами Consumer Group.
Главные триггеры: добавление/удаление консьюмера, таймаут session.timeout.ms, превышение max.poll.interval.ms.
Eager vs Cooperative: всегда используйте CooperativeStickyAssignor для исключения Stop-the-World пауз.
Static Membership: настройка group.instance.id предотвращает ребалансы при перезапуске подов в Kubernetes.
ConsumerRebalanceListener: обязателен для сброса коммитов в onPartitionsRevoked().

Различие тайм-аутов

Красные флаги на собеседовании (Чего говорить нельзя)

❌ «Ребалансировка всегда останавливает обработку всех сообщений во всей группе» — в современном кооперативном режиме останавливаются только мигрирующие партиции.
❌ «Heartbeat отправляется из метода poll()» — нет, heartbeat отправляется выделенным фоновым потоком.
❌ «Брокер сам рассчитывает распределение партиций в старом протоколе» — в классическом протоколе распределение вычисляет консьюмер-лидер, а координатор лишь утверждает его.

🟢 Junior Level

30-секундный ответ

Наглядная аналогия из жизни

Когда происходит ребалансировка?

🟡 Middle Level

Эволюция протоколов: Eager vs Cooperative Rebalancing

ConsumerRebalanceListener: фиксация позиции

Две линии защиты здоровья: Heartbeat vs Poll Interval

🔴 Senior Level

Внутренний протокол координации группы

Static Membership (group.instance.id): спасение Kubernetes

Архитектурная эволюция: KIP-848 (Next Generation Protocol)

4 Tricky Questions

1. Почему устаревший Eager Rebalancing вызывал паузы Stop-the-World для всей группы, и как Cooperative Rebalancing устраняет эту проблему?

2. Почему фоновый поток успешно шлет Heartbeat, но брокер все равно выбрасывает консьюмера из группы с CommitFailedException?

3. Что такое каскадный шторм ребалансировок (Rebalance Storm) и как разорвать этот замкнутый круг?

4. Почему нельзя хранить изменяемое не сохраняемое состояние в памяти консьюмера без реализации ConsumerRebalanceListener?

🎯 Шпаргалка для интервью

Главное о ребалансировке

Различие тайм-аутов

Красные флаги на собеседовании (Чего говорить нельзя)

Связанные вопросы

Static Membership (`group.instance.id`): спасение Kubernetes

2. Почему фоновый поток успешно шлет Heartbeat, но брокер все равно выбрасывает консьюмера из группы с `CommitFailedException`?

4. Почему нельзя хранить изменяемое не сохраняемое состояние в памяти консьюмера без реализации `ConsumerRebalanceListener`?