Как управлять событиями перед и после ребалансировки в Kafka в Spring?

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

Понимаете ли вы, что при ребалансировке партиции переходят к другим потребителям и в этот момент легко потерять или повторно обработать сообщения. В Spring Kafka для этого есть точки расширения, их ждут по именам. Следом спрашивают про хранение офсетов вне Kafka и про дубли.

Как отвечать

Начните с причин: потребитель вошёл в группу или вышел, превышен max.poll.interval.ms (по умолчанию 5 минут) из-за долгой обработки пачки, истёк session.timeout.ms (45 секунд с Kafka 3.0), изменилась подписка или число партиций.

В клиенте Kafka есть ConsumerRebalanceListener с onPartitionsRevoked, onPartitionsAssigned и onPartitionsLost. Spring Kafka добавляет ConsumerAwareRebalanceListener, где методы получают Consumer, а отзыв разбит на две фазы:

• onPartitionsRevokedBeforeCommit: до того, как контейнер закоммитит обработанные офсеты. Здесь дописывают пачку или сбрасывают буфер;
• onPartitionsRevokedAfterCommit: после коммита, для очистки локального состояния по отозванным партициям;
• onPartitionsAssigned: партиции получены, можно поднять состояние или сделать seek на офсет из внешнего хранилища;
• onPartitionsLost: партиции отобраны без нормального отзыва, коммитить здесь уже нельзя.

Подключается слушатель не через @KafkaListener, а в свойствах контейнера: factory.getContainerProperties().setConsumerRebalanceListener(listener). Для перемотки удобнее ConsumerSeekAware или наследник AbstractConsumerSeekAware.

Реже ребалансировать помогают CooperativeStickyAssignor, при котором отзываются только переезжающие партиции, и статическое членство через group.instance.id. В Kafka 4.0 стал доступен новый протокол групп (group.protocol=consumer), где назначение считает брокер.

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

Путают интерфейсы: пишут методы с параметром Consumer, реализуя обычный ConsumerRebalanceListener, где такого параметра нет.

Обрабатывают пачку дольше max.poll.interval.ms, получают бесконечные ребалансировки и лечат это ростом таймаута вместо уменьшения max.poll.records.

Не делают обработку идемпотентной. После ребалансировки часть сообщений придёт повторно, и обработчик должен это переживать.