Что такое 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`?