Как работает консьюмер в Kafka и как подписаться на топик

Что проверяют

Понимаете ли вы, что консьюмер это не просто цикл с poll(), а участник группы, которому назначают партиции и за которым следит брокер. Почти всегда следом спрашивают про коммит смещений и что будет, если обработка сообщения упала.

Как отвечать

Минимальная настройка KafkaConsumer: bootstrap.servers, group.id, десериализаторы ключа и значения. Затем выбор способа чтения:

• subscribe(topics): консьюмер входит в группу. Координатор на брокере собирает участников, партиции распределяются по partition.assignment.strategy и перераспределяются (rebalance), когда кто-то приходит или уходит. В новом протоколе group.protocol=consumer, стабильном с Kafka 4.0, распределение считает сам брокер;
• assign(partitions): партиции назначаются вручную, без ребалансов.

Внутри одной группы партицию читает не больше одного консьюмера, поэтому консьюмеров больше, чем партиций, запускать бессмысленно: лишние простаивают. Разные группы читают топик независимо.

Смещения хранятся во внутреннем топике __consumer_offsets. По умолчанию enable.auto.commit=true, коммит идёт из poll() раз в auto.commit.interval.ms (5 секунд). Для надёжности его выключают и вызывают commitSync() или commitAsync() после обработки: это at-least-once, и обработчик должен быть идемпотентным. Если сохранённого смещения нет, поведение задаёт auto.offset.reset, по умолчанию latest.

poll() нужно вызывать регулярно: если между вызовами прошло больше max.poll.interval.ms (5 минут), консьюмер считается зависшим, и его партиции уходят другим. В Spring это прячется за @KafkaListener, механика та же.

Где спотыкаются

Используют один KafkaConsumer из нескольких потоков. Он не потокобезопасен, из другого потока безопасно вызывать только wakeup(), чтобы остановить цикл.

Оставляют автокоммит и обрабатывают сообщения асинхронно. Смещение коммитится раньше, чем закончилась обработка, и при падении сообщения теряются.

Делают долгую обработку внутри цикла и получают бесконечные ребалансы из-за max.poll.interval.ms. Помогает меньший max.poll.records (по умолчанию 500) или вынос работы в отдельный пул с pause() на партициях.