Как три независимые системы могут параллельно читать из одного топика в Kafka?

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

Понимаете ли вы разницу между consumer group и отдельным потребителем. Три независимые системы должны получить каждое сообщение, а не поделить поток между собой. Следом обычно спрашивают, как масштабировать одну из систем и что будет с порядком сообщений.

Как отвечать

Ключевая мысль: у каждой системы своя consumer group, то есть свой group.id.

• группы независимы: каждая группа читает весь топик и хранит свои смещения в служебном топике __consumer_offsets. Прочитанное одной группой остаётся доступным другим, Kafka ничего не удаляет после чтения, данные живут по retention.ms;
• параллелизм внутри системы: партиции делятся между экземплярами одной группы, и в группе одну партицию читает только один потребитель. При шести партициях система поднимет до шести экземпляров, седьмой будет простаивать;
• порядок: гарантирован только внутри партиции. Если важен порядок событий заказа, ключом сообщения делают идентификатор заказа;
• своя скорость: у каждой группы свой lag, медленная аналитика не тормозит биллинг.

В Java это три KafkaConsumer с разными group.id или три @KafkaListener с разным groupId в Spring Kafka. Для новой группы важен auto.offset.reset: по умолчанию latest, и система увидит только сообщения, пришедшие после её первого подключения. Нужна история, ставьте earliest.

Хорошо добавить про ребалансировку: при добавлении экземпляра в группу партиции перераспределяются. С CooperativeStickyAssignor или новым протоколом групп из KIP-848 это проходит без остановки всех потребителей группы.

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

Ставят всем трём системам один group.id. Тогда сообщения делятся между ними, и каждая система видит лишь часть потока.

Говорят, что партиции делятся между системами: «шесть партиций, по две на систему». Это верно только для экземпляров одной группы, разные группы читают все партиции.

Обещают exactly-once по умолчанию. Коммит смещений после обработки даёт at-least-once, поэтому обработчик должен быть идемпотентным. Exactly-once требует транзакций Kafka и isolation.level=read_committed и работает в пределах самой Kafka, а не до вашей базы.