Що таке ребаланс і коли він відбувається

Ребалансування вирішує два ключові завдання:

🟢 Junior Level

30-секундна відповідь

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

Ребалансування вирішує два ключові завдання:

Масштабованість: рівномірно розподілити навантаження при додаванні нових інстансів сервісу.
Відмовостійкість: автоматично відібрати партиції у вузла, що вийшов з ладу, та передати їх живим консьюмерам.

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

Наочна аналогія з життя

Уявіть бригаду кур’єрів служби доставки:

На зміні працюють 4 кур’єри, кожному виділено свій район міста (партиція).
Якщо один кур’єр захворів і поїхав додому, диспетчер (Group Coordinator) зупиняє видачу нових замовлень на 1 хвилину, перекроює карту та ділить район кур’єра, що вибув, між трьома іншими.
Якщо на зміну вийшов п’ятий кур’єр, диспетчер забирає по одній вулиці у кожного й передає новачку.

Коли відбувається ребалансування?

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

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

3. Зміна топології топіків:
   ├── До топіка додано нові партиції
   └── Консьюмер підписаний за регулярним виразом (наприклад, "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`?