Що таке Consumer Group

Consumer Group — це бригада будівельників, що розвантажує вантажівки з будматеріалами:

🟢 Junior Level

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

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

Головні правила роботи Consumer Group:

Розподіл партицій без дублювання: кожна партиція топіка в кожен момент часу призначається суворо одному консьюмеру всередині групи.
Паралелізм: 10 партицій і 10 консьюмерів у групі дозволяють вичитувати дані в 10 паралельних потоків.
Ізольованість груп: різні групи (group.id = "billing" і group.id = "analytics") читають один і той самий топік абсолютно незалежно одна від одної. Кожна група отримує повну копію всіх повідомлень топіка зі своїм власним прогресом (зміщенням).

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

Consumer Group — це бригада будівельників, що розвантажує вантажівки з будматеріалами:

Топік — це караван із 4 вантажівок (4 партиції).
Бригада «Майстри» (group.id = "builders") складається з 4 робітників (консьюмерів).
Бригадир призначає кожному робітнику рівно по одній вантажівці. Двоє робітників не лізуть в одну машину — вони не заважають один одному і не вивантажують одну й ту саму цеглу двічі.
Якщо на об’єкті паралельно працює бригада аудиторів (group.id = "auditors"), вони перевіряють ті самі вантажівки незалежно від будівельників, не забираючи в них матеріали.

Базовий приклад налаштування в Spring Boot 3

@Service
public class OrderNotificationConsumer {

    // Група "notifications-group": кілька реплік сервісу поділять партиції топіка
    @KafkaListener(
        topics = "orders", 
        groupId = "notifications-group",
        concurrency = "3" // Запуск 3 паралельних потоків-консьюмерів всередині JVM
    )
    public void consumeOrder(ConsumerRecord<String, String> record) {
        System.out.printf("Партиція: %d, Offset: %d, Повідомлення: %s%n",
            record.partition(), record.offset(), record.value());
    }
}

🟡 Middle Level

Співвідношення кількості консьюмерів і партицій

Поведінка групи жорстко залежить від співвідношення кількості інстансів і партицій:

Сценарій А: 4 партиції, 2 консьюмери (Consumers < Partitions)
  [Partition 0] ──┐
  [Partition 1] ──┴──► [ Consumer 1 ]
  [Partition 2] ──┐
  [Partition 3] ──┴──► [ Consumer 2 ]

Сценарій Б: 4 партиції, 4 консьюмери (Consumers == Partitions) — Ідеальний баланс!
  [Partition 0] ─────► [ Consumer 1 ]
  [Partition 1] ─────► [ Consumer 2 ]
  [Partition 2] ─────► [ Consumer 3 ]
  [Partition 3] ─────► [ Consumer 4 ]

Сценарій В: 4 партиції, 6 консьюмерів (Consumers > Partitions)
  [Partition 0] ─────► [ Consumer 1 ]
  [Partition 1] ─────► [ Consumer 2 ]
  [Partition 2] ─────► [ Consumer 3 ]
  [Partition 3] ─────► [ Consumer 4 ]
                       [ Consumer 5 ] ──► IDLE (Простоює в холостому режимі)
                       [ Consumer 6 ] ──► IDLE (Простоює в холостому режимі)

Якщо консьюмерів більше, ніж партицій — «зайві» інстанси не отримують повідомлень і знаходяться в режимі очікування (standby/idle).
Вони підключаться до роботи лише в тому разі, якщо один з активних консьюмерів вийде з ладу.

Де зберігаються зміщення: внутрішній топік `__consumer_offsets`

Раніше (у старих версіях Kafka до 0.9) зміщення зберігалися в ZooKeeper. Це перевантажувало ZK і призводило до збоїв. Сьогодні Kafka зберігає всі підтверджені офсети у внутрішньому компактному системному топіку __consumer_offsets:

Топік складається з 50 партицій із фактором реплікації 3.
Формат ключа запису: [group.id, topic, partition].
Формат значення: [offset, metadata, commit_timestamp].
Політика очищення: cleanup.policy = compact (на диску зберігається лише найостанніший підтверджений офсет для кожної конкретної пари «група-партиція»).

Що таке Rebalance (Ребалансування)?

Ребалансування — це розподілений процес перерозподілу партицій між членами Consumer Group. Воно запускається брокером у таких випадках:

До групи підключився новий консьюмер (масштабування сервісу вгору).
Наявний консьюмер відключився або впав через збій (масштабування вниз або краш пода).
Адміністратор додав нові партиції в топік.
Консьюмер не встиг викликати poll() протягом заданого таймауту (max.poll.interval.ms).

🔴 Senior Level

Протоколи ребалансування: Eager vs Cooperative Sticky Rebalance

1. Застарілий протокол Eager Rebalance (Stop-the-World):

При будь-якій зміні у складі групи ВСІ консьюмери повністю відкликають свої поточні партиції та зупиняють вичитування.
Весь кластер споживачів завмирає на 5–30 секунд (Stop-the-World).
Обчислюється новий розподіл, і консьюмери заново підключаються до партицій.
У Kubernetes при rolling update (коли по черзі перезапускаються 20 подів) Eager Rebalance викликав «шторм ребалансувань» і тривалі простої сервісів!

2. Сучасний Cooperative Sticky Rebalance (Kafka 2.3+ / KIP-429):

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

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

Час простою скорочується до мілісекунд:

# Налаштування сучасного кооперативного assignor
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Static Membership (KIP-345): порятунок для Kubernetes

При розгортанні в Kubernetes поди постійно перезапускаються (rolling updates, canary, міграція вузлів K8s). При динамічному членстві кожен перезапуск пода генерує 2 ребаланси:

Под зупинився $\to$ виключення з групи $\to$ ребаланс.
Новий под піднявся $\to$ додавання до групи $\to$ ребаланс.

Рішення — Static Membership: Задання унікального статичного ідентифікатора пода:

group.instance.id=payment-service-pod-0

Коли под перезавантажується, брокер не запускає ребаланс негайно.
Координатор групи утримує призначені за цим подом партиції протягом інтервалу session.timeout.ms (наприклад, 45 секунд).
Якщо новий под з тим самим group.instance.id повертається протягом цього часу, він миттєво забирає свої партиції без запуску ребалансування всієї групи!

Координатор групи (Group Coordinator)

Як визначається, який саме брокер керує вашою Consumer Group:

Обчислюється номер партиції системного топіка зміщень: $\text{partition} = \text{Utils.abs}(\text{groupId.hashCode()}) \pmod{50}$
Брокер, який є лідером обчисленої партиції топіка __consumer_offsets, автоматично стає Group Coordinator для цієї групи.
Усі консьюмери групи надсилають запити JoinGroup, SyncGroup, Heartbeat і коміти зміщень суворо на цей брокер.

4 Tricky Questions

1. Чому Eager Rebalance призводив до каскадних аварій у Kubernetes, і як Cooperative Sticky Assignor вирішує цю проблему?

Відповідь:

В Eager Rebalance при виході з ладу або рестарті одного інстанса координатор відкликає абсолютно ВСІ партиції у ВСІХ учасників групи. Усі потоки зупиняють обробку даних на час протоколу Join/Sync.
У Kubernetes при послідовному рестарті 10 подів (Rolling Update) запускалося 20 послідовних фаз Stop-the-World. Накопичений за час пауз лаг перевантажував процесори подів, збільшувався час GC, сервіси пропускали таймаути max.poll.interval.ms і вибивалися з групи знову, викликаючи нескінченний самопідтримуваний «шторм ребалансувань» (Rebalance Storm).
Cooperative Sticky Rebalance усуває глобальний Stop-the-World: інстанси, чиї партиції не зачеплені змінами, взагалі не припиняють вичитування та обробку повідомлень. Мігрують лише звільнені партиції, що звело час перерозподілу до часток секунди.

2. Чому консьюмер може бути викинутий із Consumer Group за таймаутом, якщо його фоновий потік Heartbeat Thread працює ідеально та справно шле пінги брокеру?

Відповідь:

Починаючи з версії Kafka 0.10.1, відправлення сигналів життя (Heartbeat) та обробка повідомлень рознесені по різних потоках:
- Фоновий потік шле пінги брокеру кожні heartbeat.interval.ms і перевіряє session.timeout.ms.
- Основний потік застосунку послідовно викликає consumer.poll() і передає пачку повідомлень у бізнес-обробку.
Якщо бізнес-логіка (наприклад, важкий SQL-запит, зовнішній HTTP-виклик або блокування) зависає довше, ніж значення конфігурації max.poll.interval.ms (за замовчуванням 5 хвилин):
Координатор групи вважає, що консьюмер завис у дедлоку (livelock/stuck). Попри те, що heartbeat-потік живий, брокер примусово виганяє консьюмера з групи та запускає ребалансування.
Коли завислий консьюмер нарешті «відмирає» і намагається зафіксувати офсети, він отримує виняток CommitFailedException.

3. Як саме визначається брокер, що виступає в ролі Group Coordinator для вашої групи, і чому це важливо для розподілу навантаження?

Відповідь:

Визначення координатора суворо математичне: береться ім’я групи group.id, обчислюється Math.abs(groupId.hashCode()) % offsets.topic.num.partitions (за замовчуванням 50).
Отриманий номер (наприклад, 14) вказує на конкретну партицію внутрішнього топіка __consumer_offsets-14.
Той брокер кластера, який на даний момент є лідером репліки партиції 14, стає персональним координатором для цієї групи.
Практична важливість: якщо у вас у кластері тисячі мікросервісів, важливо, щоб імена їхніх group.id були різноманітними та рівномірно хешувалися по всіх 50 партиціях __consumer_offsets. Якщо всі команди назвуть свої групи зі спільним префіксом, який випадково потрапить в одну партицію, один брокер виявиться перевантаженим координацією всіх консьюмерів компанії.

4. Що станеться, якщо под із налаштованим `group.instance.id` аварійно впав (OOMKilled) і не зміг перезапуститися протягом часу `session.timeout.ms`?

Відповідь:

Якщо ввімкнено Static Membership, брокер при відключенні пода не відбирає його партиції одразу, а чекає на закінчення інтервалу session.timeout.ms (наприклад, 45 секунд).
Протягом цих 45 секунд партиції впалого пода «висять у повітрі» — їх ніхто не читає, повідомлення накопичуються в лозі, лаг зростає.
Якщо за 45 секунд под не повернувся до ладу, координатор фіксує факт аварії, офіційно оголошує учасника мертвим, виключає його зі складу групи та запускає процедуру перерозподілу партицій між рештою живих подів.

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

Головне про Consumer Group

Сутність: об’єднання споживачів для паралельного читання топіка.
Головний інваріант: 1 партиція топіка $\leftrightarrow$ максимум 1 консьюмер всередині групи.
Зайві консьюмери: якщо споживачів більше, ніж партицій — вони простоюють в idle.
Різні групи: читають топік незалежно, реалізуючи класичну модель Publish-Subscribe.
Зберігання зміщень: у внутрішньому топіку __consumer_offsets (compacted, 50 партицій).
Координатор: брокер-лідер партиції hash(groupId) % 50.
Cooperative Sticky Assignor: сучасний стандарт без Stop-the-World пауз.
Static Membership (group.instance.id): запобігає ребалансам при перезапусках подів у Kubernetes.
Два різні таймаути: session.timeout.ms (для heartbeat) та max.poll.interval.ms (для циклу poll()).

Порівняння протоколів ребалансування

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

❌ «Два консьюмери з однієї групи можуть одночасно вичитувати одну партицію для прискорення» — груба помилка, це суворо заборонено моделлю Kafka.
❌ «Офсети консьюмерів досі зберігаються в ZooKeeper» — вони зберігаються в топіку __consumer_offsets уже багато років.
❌ «Якщо heartbeat працює, консьюмера ніколи не виженуть із групи» — якщо перевищити max.poll.interval.ms, брокер вижене консьюмера незалежно від heartbeat.
❌ «Автокоміт офсетів — найкращий вибір для надійного продакшену» — для надійності потрібен ручний коміт після успішної обробки бізнес-транзакції.

🟢 Junior Level

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

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

Базовий приклад налаштування в Spring Boot 3

🟡 Middle Level

Співвідношення кількості консьюмерів і партицій

Де зберігаються зміщення: внутрішній топік __consumer_offsets

Що таке Rebalance (Ребалансування)?

🔴 Senior Level

Протоколи ребалансування: Eager vs Cooperative Sticky Rebalance

1. Застарілий протокол Eager Rebalance (Stop-the-World):

2. Сучасний Cooperative Sticky Rebalance (Kafka 2.3+ / KIP-429):

Static Membership (KIP-345): порятунок для Kubernetes

Координатор групи (Group Coordinator)

4 Tricky Questions

1. Чому Eager Rebalance призводив до каскадних аварій у Kubernetes, і як Cooperative Sticky Assignor вирішує цю проблему?

2. Чому консьюмер може бути викинутий із Consumer Group за таймаутом, якщо його фоновий потік Heartbeat Thread працює ідеально та справно шле пінги брокеру?

3. Як саме визначається брокер, що виступає в ролі Group Coordinator для вашої групи, і чому це важливо для розподілу навантаження?

4. Що станеться, якщо под із налаштованим group.instance.id аварійно впав (OOMKilled) і не зміг перезапуститися протягом часу session.timeout.ms?

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

Головне про Consumer Group

Порівняння протоколів ребалансування

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

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

Де зберігаються зміщення: внутрішній топік `__consumer_offsets`

4. Що станеться, якщо под із налаштованим `group.instance.id` аварійно впав (OOMKilled) і не зміг перезапуститися протягом часу `session.timeout.ms`?