📨 Раздел 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.

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