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

Що станеться при додаванні нового консьюмера в групі

При підключенні нового консьюмера до вже працюючої Consumer Group в Apache Kafka запускається процес ребалансування (Rebalance).

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

🟢 Junior Level

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

При підключенні нового консьюмера до вже працюючої Consumer Group в Apache Kafka запускається процес ребалансування (Rebalance).

Що відбувається за кроками:

  1. Реєстрація: новий інстанс відправляє брокеру (Group Coordinator) запит на вступ до групи.
  2. Перерозподіл: координатор групи перераховує розподіл партицій, щоб виділити новачку частину роботи.
  3. Відкликання партицій: у деяких раніше працюючих консьюмерів відкликається частина партицій і передається новому учаснику.
  4. Відновлення вичитування: новий консьюмер починає читати призначені йому партиції зі зміщень, зафіксованих у топіку __consumer_offsets.

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

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

  • У місті 6 районів (6 партицій).
  • На зміні працюють 2 кур’єри: Кур’єр А розвозить замовлення по районах 1, 2 і 3, а Кур’єр Б — по районах 4, 5 і 6.
  • В офіс приходить новий співробітник (Кур’єр В). Бригадир оголошує коротку п’ятихвилинку:
    • Кур’єр А віддає район 3 новенькому.
    • Кур’єр Б віддає район 6 новенькому.
  • Тепер навантаження розподілено порівну: у кожного по 2 райони. Усі кур’єри працюють швидше, доставка прискорюється!

Схема розподілу до і після додавання

Було (2 консьюмери на 4 партиції):
  [Partition 0] ──┐
  [Partition 1] ──┴──► [ Consumer 1 ]
  [Partition 2] ──┐
  [Partition 3] ──┴──► [ Consumer 2 ]

Підключився Consumer 3 ──► [ REBALANCE ] ──► Стало (3 консьюмери на 4 партиції):
  [Partition 0] ──┐
  [Partition 1] ──┴──► [ Consumer 1 ]
  [Partition 2] ──────► [ Consumer 2 ]
  [Partition 3] ──────► [ Consumer 3 ]  <-- Новачок забрав партицію 3

🟡 Middle Level

Покроковий протокол підключення до групи

[ Новий Консьюмер C3 ]            [ Group Coordinator (Брокер) ]         [ Працюючі C1 і C2 ]
          │                                      │                                 │
          │ 1. JoinGroup (memberId="")           │                                 │
          ├─────────────────────────────────────►│                                 │
          │ 2. Помилка: MEMBER_ID_REQUIRED       │                                 │
          │◄─────────────────────────────────────┤                                 │
          │                                      │                                 │
          │ 3. JoinGroup (з виданим memberId)    │                                 │
          ├─────────────────────────────────────►│                                 │
          │                                      │ 4. Coordinator чекає завершення │
          │                                      │    поточного poll() у C1 і C2   │
          │                                      │    (в межах max.poll.interval)  │
          │                                      │◄────────────────────────────────┤
          │                                      │    JoinGroup від C1 і C2        │
          │ 5. Призначення Лідера (зазвичай C1)  │                                 │
          │ 6. Лідер рахує Assignor.assign()     │                                 │
          │ 7. SyncGroup Request                 │ 7. SyncGroup Request            │
          ├─────────────────────────────────────►│◄────────────────────────────────┤
          │ 8. SyncGroup Response (Новий план)   │ 8. SyncGroup Response           │
          │◄─────────────────────────────────────┼────────────────────────────────►│
          │  C3 отримує Partition 3              │  C2 віддала P3, залишилась з P2 │

Роль ConsumerRebalanceListener при відкликанні партиції

Коли брокер повідомляє консьюмеру, що в нього забирають партицію для передачі новому учаснику, викликається метод onPartitionsRevoked(). Критично важливо коректно обробити цю подію:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("Партиції відкликаються брокером: {}", partitions);
        // 1. Дочікуємося обробки поточного батча повідомлень
        // 2. Синхронно фіксуємо зміщення, щоб новачок не прочитав дублікати!
        consumer.commitSync();
        // 3. Скидаємо проміжні буфери в БД
        flushInternalBuffers(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Призначено нові партиції: {}", partitions);
        // Ініціалізуємо локальні кеші або виконуємо consumer.seek()
    }
});

🔴 Senior Level

Чому в CooperativeStickyAssignor додавання консьюмера вимагає ДВОХ раундів ребалансування?

У застарілому протоколі Eager Rebalance всі консьюмери здавали всі свої партиції за 1 раунд із повною паузою (Stop-the-World). У сучасному CooperativeStickyAssignor (KIP-429) ребалансування розбито на два швидких кооперативних кроки, щоб ніхто не зупиняв читання незачеплених партицій:

РАУНД 1: Визначення та відкликання мігруючих партицій
  - C3 підключається. C1 і C2 продовжують вичитувати дані!
  - Лідер бачить: C1 тримає {P0, P1}, C2 тримає {P2, P3}.
  - Лідер вирішує: P3 потрібно віддати новачку C3.
  - Рішення 1-го раунду: C1 залишає {P0, P1}, C2 відкликає P3 (залишається з P2), C3 поки отримує ПОРОЖНІЙ набір {}.
  - C2 викликає onPartitionsRevoked({P3}), фіксує зміщення і відпускає партицію. C1 ні на секунду не зупиняється!

РАУНД 2: Передача звільненої партиції
  - Координатор фіксує, що P3 звільнилася.
  - Негайно ініціюється короткий другий раунд:
  - C3 отримує {P3}, C1 зберігає {P0, P1}, C2 зберігає {P2}.
  - Вичитування повністю збалансоване без жодної глобальної паузи!

Проблема зомбі-консьюмерів (Zombie Fencing & Epochs)

Що коли вузол C2 завис у важкому GC, брокер визнав його мертвим, віддав партицію P3 новачку C3, а потім C2 «відвис» і намагається дописати щось у БД або закомітити офсет?

  1. Split-Brain Risk: два консьюмери одночасно пишуть дані однієї й тієї ж партиції в базу даних.
  2. Захист поколіннями (Generation ID / Member Epoch):
    • При кожній зміні складу групи брокер збільшує номер покоління Generation ID (у KIP-848 — Member Epoch).
    • Будь-який мережевий запит до брокера (OffsetCommitRequest, FetchRequest) містить поточний Generation ID.
    • Якщо брокер бачить, що Generation ID консьюмера C2 застарів, запит негайно відхиляється з помилкою CommitFailedException або FencedMemberEpochException.

4 Tricky Questions

1. Чому при найпершому вступі до групи новий консьюмер надсилає запит JoinGroup ДВІЧІ?

Відповідь:

  • У ранніх версіях Kafka консьюмер генерував свій member.id сам або брокер створював його одразу на старті ребалансу. Це породжувало вразливість: при рестарті консьюмерів брокер ініціював хибні ребаланси на фантомних учасників.
  • Починаючи з KIP-394, запроваджено суворий двофазний handshake:
    1. Новий консьюмер надсилає перший JoinGroupRequest із порожнім ідентифікатором member.id = "".
    2. Брокер не починає ребалансування! Він лише реєструє намір, генерує унікальний ID (наприклад, consumer-group-1-UUID) і повертає помилку MEMBER_ID_REQUIRED із прикріпленим згенерованим ID.
    3. Клієнт зберігає отриманий ID і надсилає другий JoinGroupRequest, що вже містить валідний member.id.
  • Лише при отриманні другого запиту координатор офіційно включає учасника до групи та запускає процедуру перерозподілу. Це запобігає випадковим ребалансам через «мертві» мережеві запити.

2. Що станеться, якщо в момент додавання нового консьюмера працюючий інстанс виконує довгу бізнес-транзакцію (наприклад, 15 секунд), а max.poll.interval.ms дорівнює 30 секундам?

Відповідь:

  • Брокер не може силою перервати виконання Java-коду всередині методу consume(). Він лише встановлює прапорець RebalanceInProgress у відповідях на Heartbeat.
  • Координатор групи буде терпляче чекати, доки працюючий інстанс закінчить свою 15-секундну транзакцію та прийде на наступну ітерацію consumer.poll().
  • Щойно інстанс викликає poll(), клієнтська бібліотека Kafka бачить прапорець ребалансу, викликає ConsumerRebalanceListener.onPartitionsRevoked() і шле свій JoinGroup.
  • Час початку фактичного перерозподілу партицій затягується рівно на ці 15 секунд. Усі інші консьюмери (включаючи новачка) чекатимуть закінчення цієї транзакції у фазі JoinGroup.

3. Новий консьюмер підключився до групи, ребалансування завершилося, йому призначено партицію 2. З якого саме зміщення він почне читати повідомлення?

Відповідь:

  • Логіка визначення стартового офсету суворо регламентована:
    1. Консьюмер надсилає брокеру запит OffsetFetchRequest для призначеної партиції 2.
    2. Брокер читає останнє підтверджене зміщення для цієї групи з внутрішнього топіка __consumer_offsets.
    3. Якщо в __consumer_offsets знайдено запис (наприклад, офсет 45100), новий консьюмер починає вичитування суворо зі зміщення 45100.
    4. Якщо зміщення в __consumer_offsets немає (топік читається цією групою вперше або термін зберігання офсетів offsets.retention.minutes минув): консьюмер дивиться на свій локальний параметр auto.offset.reset:
      • earliest: почне читати із самого початку логу партиції (офсет 0 або найстаріший доступний на диску).
      • latest: переміститься в самий кінець логу і чекатиме лише нових повідомлень.
      • none: викине фатальний виняток NoOffsetForPartitionException.

4. Що станеться, якщо новий консьюмер вступив у групу, але кількість консьюмерів перевищила кількість партицій топіка (наприклад, 5-й консьюмер на топік з 4 партиціями)?

Відповідь:

  • Процедура ребалансування виконається штатно.
  • Стратегія розподілу (PartitionAssignor) розподілить 4 наявні партиції між будь-якими 4 учасниками групи.
  • П’ятий консьюмер отримає порожній список призначень: assignedPartitions = [].
  • Він перейде в режим Idle:
    • Він продовжить надсилати Heartbeat координатору кожні heartbeat.interval.ms, підтверджуючи свою життєздатність.
    • Метод consumer.poll() миттєво повертатиме порожній список записів ConsumerRecords.empty().
    • Жодного навантаження з бізнес-логіки він не несе і чекає, доки хтось із 4 активних учасників впаде або залишить групу.

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

Головне про додавання нового консьюмера

  • Що відбувається: Запускається Rebalance для рівномірного перерозподілу партицій.
  • Двофазний handshake (KIP-394): Перший JoinGroup отримує member.id, другий запускає ребаланс.
  • Старий Eager протокол: Повна зупинка всіх учасників групи (Stop-the-World на 5–30 секунд).
  • Сучасний Cooperative Sticky (KIP-429): Проходить у 2 коротких етапи, незачеплені партиції читаються без зупинки.
  • Обов’язок розробника: Реалізувати ConsumerRebalanceListener.onPartitionsRevoked() і викликати commitSync().
  • З якого офсету читає новачок: З останнього закоміченого зміщення з топіка __consumer_offsets.
  • Захист від зомбі: Generation ID / Member Epoch блокує застарілі інстанси з помилкою CommitFailedException.

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

| Характеристика | Eager Assignor | CooperativeStickyAssignor | | :— | :— | :— | | Зупинка роботи | Усі інстанси зупиняють читання | Читання продовжується на незачеплених партиціях | | Кількість раундів | 1 раунд (із повною паузою) | 2 швидких раунди (відкликання $\to$ призначення) | | Поведінка при зростанні подів | Шторм ребалансувань, сплеск лагу | М’який інкрементальний перерозподіл |

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

  • ❌ «Новий консьюмер одразу ж починає читати повідомлення паралельно зі старим консьюмером» — читання почнеться лише ПІСЛЯ завершення ребалансу і відкликання партиції у старого.
  • ❌ «Новачок завжди читає повідомлення із самого початку топіка (з 0)» — він починає з останнього закоміченого офсету групи в __consumer_offsets.
  • ❌ «Ребаланс миттєво перериває виконання потоку на працюючому сервері» — брокер чекає завершення поточної ітерації poll() у межах max.poll.interval.ms.

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