📨 Розділ 15 · Питання #6

Як працює балансування консьюмерів у групі

Балансування — це розподіл авіарейсів між авіадиспетчерами:

📨 Kafka 📄 Вихідний Markdown

🟢 Junior Level

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

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

Навіщо воно потрібне:

  1. Автоматизація: розробнику не потрібно вручну прописувати, який сервер яку партицію читає.
  2. Еластичність: при падінні або додаванні нових інстансів мікросервісу брокер ініціює ребалансування (Rebalance) і справедливо перерозподіляє партиції між рештою живих учасників.
  3. Суворий інваріант: всередині групи одна партиція призначається суворо одному консьюмеру. При цьому один консьюмер може читати відразу кілька партицій.

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

Балансування — це розподіл авіарейсів між авіадиспетчерами:

  • В аеропорту 6 злітно-посадкових смуг (6 партицій топіка).
  • На зміну заступили 3 авіадиспетчери (3 консьюмери в групі).
  • Старший диспетчер виділяє кожному рівно по 2 смуги: Диспетчер А стежить за смугами 0 і 1, Диспетчер Б — за 2 і 3, Диспетчер В — за 4 і 5.
  • Якщо Диспетчер В раптово втрачає свідомість (краш сервісу), старший диспетчер негайно оголошує перерозподіл і передає смуги 4 і 5 диспетчерам А і Б, що залишилися. Жодна смуга не залишається без контролю!

Базовий приклад структури розподілу

[ Топік: 6 партицій ]   P0    P1    P2    P3    P4    P5
                         │     │     │     │     │     │
                         └──┬──┘     └──┬──┘     └──┬──┘
                            ▼           ▼           ▼
                      [Consumer A] [Consumer B] [Consumer C]
                      (2 партиції) (2 партиції) (2 партиції)

У коді Java консьюмер просто вказує group.id і підписується на топік:

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "orders-processing-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders")); // Kafka сама розподілить партиції

🟡 Middle Level

4 основні стратегії розподілу (PartitionAssignor)

Вибір стратегії задається параметром partition.assignment.strategy:

Стратегія Принцип роботи Плюси Мінуси
RangeAssignor (історичний дефолт) Розглядає кожен топік ізольовано і ділить партиції на неперервні діапазони. Проста реалізація. Колосальний перекіс навантаження при підписці на кілька топіків!
RoundRobinAssignor Збирає всі партиції всіх топіків в один список і роздає по колу консьюмерам. Рівномірна кількість партицій у всіх учасників. Повністю перемішує партиції при будь-якому ребалансі.
StickyAssignor Зберігає максимальну кількість поточних призначень партицій при ребалансі. Мінімізує міграцію партицій, береже кеш даних. Використовує застарілий Eager протокол (Stop-the-World).
CooperativeStickyAssignor (сучасний стандарт) Інкрементальне ребалансування без зупинки незачеплених партицій. 🚀 Мінімальний час простою, відсутність глобального Stop-the-World. Доступний з Kafka 2.3+.

Проблема перекосу RangeAssignor (Range Skew)

Уявіть, що мікросервіс підписаний на 10 топіків, і в кожному топіку по 3 партиції (P0, P1, P2). Усього в групі 2 консьюмери (C1 та C2):

  • Формула RangeAssignor для кожного топіка: $3 \div 2 = 1.5 \to$ Консьюмер C1 отримує 2 партиції (P0, P1), а Консьюмер C2 отримує 1 партицію (P2).
  • Цей розрахунок повторюється для кожного з 10 топіків!
  • У підсумку: Консьюмер C1 отримує 20 партицій, а Консьюмер C2 — лише 10 партицій!
  • Консьюмер C1 перевантажений удвічі сильніше, впирається в CPU, накопичує лаг, а C2 недовантажений.
  • Висновок: ніколи не використовуйте RangeAssignor, якщо група читає кілька топіків!

Роль ConsumerRebalanceListener

При виникненні ребалансування консьюмер повинен встигнути коректно зберегти свій стан:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Викликається ДО того, як брокер відбере партиції.
        // ОБОВ'ЯЗКОВО: синхронно фіксуємо зміщення, щоб інший вузол не обробив дублі!
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Викликається ПІСЛЯ того, як брокер призначив консьюмеру нові партиції.
        // Ініціалізуємо локальні кеші, зміщення в БД або метрики.
    }
});

🔴 Senior Level

Анатомія класичного протоколу ребалансування (JoinGroup / SyncGroup)

Багато хто помилково вважає, що сам брокер Kafka розподіляє партиції між клієнтами. Це не так! У класичному протоколі Kafka розподіл виконує один із клієнтських консьюмерів (Leader):

[ Consumer A ]               [ Group Coordinator (Брокер) ]               [ Consumer B ]
      │                                    │                                    │
      │ 1. JoinGroup Request               │                                    │
      ├───────────────────────────────────►│ 1. JoinGroup Request               │
      │                                    │◄───────────────────────────────────┤
      │                                    │                                    │
      │ (Брокер обирає Лідером Client A)   │                                    │
      │                                    │                                    │
      │ 2. JoinGroup Response              │ 2. JoinGroup Response              │
      │ (allMembers + YOU ARE LEADER)      │ (memberId + LEADER IS A)           │
      │◄───────────────────────────────────┼───────────────────────────────────►│
      │                                    │                                    │
      │ [ Лідер A локально виконує         │                                    │
      │   PartitionAssignor.assign() ]     │                                    │
      │                                    │                                    │
      │ 3. SyncGroup Request               │ 3. SyncGroup Request               │
      │ (відправляє готовий план для всіх) │ (порожній запит очікування)        │
      ├───────────────────────────────────►│◄───────────────────────────────────┤
      │                                    │                                    │
      │ 4. SyncGroup Response              │ 4. SyncGroup Response              │
      │ (отримує свої партиції: P0, P1)    │ (отримує свої партиції: P2)        │
      │◄───────────────────────────────────┼───────────────────────────────────►│

KIP-848: Революція протоколу ребалансування (Kafka 3.7+ / 4.0)

Класичний протокол (KIP-429 і нижче) мав архітектурний недолік: залежність від клієнта-лідера та двофазне опитування JoinGroup $\to$ SyncGroup. У межах KIP-848 (Next Generation Consumer Protocol):

  1. Server-Side Assignment: логіку обчислення розподілу перенесено повністю на сторону Group Coordinator (Брокера).
  2. Heartbeat-Driven Reconciliation: зникли блокуючі фази JoinGroup/SyncGroup. Консьюмер просто надсилає регулярний RPC-запит ConsumerGroupHeartbeat.
  3. Брокер у тілі відповіді Heartbeat інкрементально повертає директиви: «Віддай партицію 2, візьми партицію 5». Ребаланс став повністю безшовним і прозорим.

Катастрофічна петля HPA (Kubernetes Autoscaling Death Loop)

Поширений антипатерн в архітектурі хмарних мікросервісів:

  1. DevOps налаштовують Kubernetes Horizontal Pod Autoscaler (HPA) на основі утилізації процесора (targetCPUUtilizationPercentage: 70%).
  2. У топік надходить сплеск повідомлень. Навантаження на поди консьюмерів зростає до 85% CPU.
  3. HPA бачить перевантаження і піднімає 3 нових поди.
  4. Нові поди вступають у Consumer Group, що ініціює ребалансування.
  5. При ребалансуванні потоки зупиняються, фіксують офсети, переініціалізують з’єднання. Мережеві стеки та GC короткочасно завантажують процесор на 100%.
  6. HPA фіксує зростання навантаження до 95% CPU і піднімає ще 5 подів!
  7. Починається безперервний шторм ребалансувань, обробка повідомлень повністю блокується, лаг летить у мільйони записів.
    • Рішення: масштабувати Kafka-консьюмери в Kubernetes лише за бізнес-метрикою Consumer Lag (через KEDA або Prometheus Adapter) і суворо із захисним інтервалом охолодження (cooldown period не менше 5–10 хвилин).

4 Tricky Questions

1. Чому RangeAssignor призводить до катастрофічного перекосу навантаження при вичитуванні кількох топіків, і чому його досі не видалили?

Відповідь:

  • RangeAssignor групує партиції кожного топіка суворо за числовим порядком і ділить їх на кількість консьюмерів незалежними діапазонами.
  • Якщо у вас є 5 топіків по 3 партиції (0, 1, 2) і 2 консьюмери (C1 та C2), ділення $3 / 2 = 1.5$ округлюється на користь першого консьюмера. У кожному з 5 топіків консьюмер C1 отримує партиції 0 і 1, а C2 — партицію 2.
  • У сумі C1 читатиме 10 партицій, а C2 — лише 5 партицій (вдвічі менше!). Якщо топіків буде 50, C1 отримає 100 партицій, а C2 — 50.
  • Причиною, чому він залишався дефолтним довгі роки, була зворотна сумісність (Colocated Partitions): якщо продюсер пише пов’язані дані в топіки A і B за одним ключем в однакові номери партицій, RangeAssignor гарантував, що партиція 0 топіка A і партиція 0 топіка B потраплять на один і той самий інстанс консьюмера (зручно для локальних JOIN у ранніх версіях).

2. Хто фізично виконує Java-код стратегії розподілу партицій (PartitionAssignor.assign()) — брокер Kafka чи клієнт?

Відповідь:

  • У класичному протоколі Kafka (до KIP-848) алгоритм assign() виконується на одному з клієнтських інстансів консьюмерів, обраному як Лідер групи (Group Leader).
  • Брокер (Group Coordinator) збирає список усіх учасників групи через запити JoinGroup і передає цей список першому консьюмеру, що підключився (Лідеру).
  • Лідер локально у своєму JVM-процесі запускає Java-код налаштованого класу PartitionAssignor, формує Map<MemberId, List<TopicPartition>> і надсилає цей результат назад брокеру в запиті SyncGroup.
  • І лише починаючи з KIP-848 (Kafka 3.7+ / 4.0) брокер бере розрахунок призначень на себе (Server-Side Assignor).

3. Що станеться, якщо в методі ConsumerRebalanceListener.onPartitionsRevoked() виникне необроблений RuntimeException (наприклад, таймаут БД)?

Відповідь:

  • Якщо метод onPartitionsRevoked() викидає виняток, консьюмер перериває фазу підготовки до ребалансу.
  • Незбережені зміщення оброблених записів не будуть зафіксовані в __consumer_offsets.
  • Сам інстанс консьюмера визнає внутрішній стан неузгодженим, перериває цикл poll() і закриває клієнтську сесію з викиданням критичної помилки KafkaException.
  • Партиції відкликаються брокером примусово за таймаутом. Новий консьюмер, якому брокер перепризначить ці партиції, почне вичитування зі старих закомічених офсетів, що неминуче призведе до повторної обробки пачки повідомлень (Duplicate Processing).

4. Чим принципово відрізняється поведінка StickyAssignor від CooperativeStickyAssignor при додаванні нового інстанса до працюючої групи?

Відповідь:

  • StickyAssignor використовує протокол Eager: коли новий учасник входить до групи, брокер вимагає від ВСІХ працюючих учасників негайно припинити роботу та здати ВСІ свої партиції. Вся група стає на паузу (Stop-the-World). Потім StickyAssignor призначає більшості подів їхні старі партиції, а частину партицій віддає новачку. Пауза зачіпає 100% учасників.
  • CooperativeStickyAssignor використовує протокол Cooperative (KIP-429): при додаванні новачка працюючі консьюмери НЕ зупиняють вичитування своїх партицій. На першому кроці брокер обчислює, які саме 2–3 партиції потрібно забрати для новачка, і просить віддати лише їх. Усі інші 20 консьюмерів продовжують безперервно обробляти трафік без найменшої паузи!

🎯 Шпаргалка для інтерв’ю

Головне про балансування

  • Сутність: автоматичний розподіл партицій топіка між консьюмерами групи.
  • Інваріант: 1 партиція = суворо 1 консьюмер групи в будь-який момент часу.
  • Стратегія для Production: CooperativeStickyAssignor (мінімальний простій, інкрементальне перенесення).
  • Небезпека RangeAssignor: катастрофічний перекіс партицій при підписці на кілька топіків.
  • Хто рахує призначення: у класичному протоколі — клієнтський консьюмер-Лідер групи (не брокер!). У KIP-848 — брокер.
  • ConsumerRebalanceListener: обов’язковий для синхронного коміту офсетів в onPartitionsRevoked().
  • HPA в Kubernetes: масштабувати поди консьюмерів за CPU заборонено (викликає шторм ребалансувань); масштабувати лише за Consumer Lag з cooldown period.

Порівняння стратегій PartitionAssignor

| Стратегія | Протокол | Рівномірність | Переміщення партицій | | :— | :— | :— | :— | | RangeAssignor | Eager (Stop-the-World) | ❌ Дуже погана на multi-topic | Середнє | | RoundRobinAssignor | Eager (Stop-the-World) | ✅ Рівномірна | ❌ Перемішує все поспіль | | StickyAssignor | Eager (Stop-the-World) | ✅ Рівномірна | ✅ Мінімальне переміщення | | CooperativeStickyAssignor | Cooperative (Non-blocking) | ✅ Рівномірна | 🚀 Інкрементальне без пауз |

Червоні прапорці на співбесіді (Чого говорити не можна)

  • ❌ «Брокер Kafka сам вирішує в Java-коді, кому яку партицію віддати» — у стандартній Kafka розподіл обчислює клієнт-Лідер групи.
  • ❌ «RangeAssignor завжди ділить партиції порівну» — при непарних партиціях на кількох топіках перекіс сягає сотень відсотків.
  • ❌ «При ребалансуванні вся система завжди зупиняється на 30 секунд» — з CooperativeStickyAssignor зупинка відбувається інкрементально за мілісекунди.

Пов’язані запитання