Как @KafkaListener связан с listener container и offset commit?
- Когда Spring Kafka коммитит offset?
@KafkaListener — метод, который вызывает listener container: контейнер крутит цикл poll() в своих потоках, вызывает метод для записей и коммитит offset по AckMode. По умолчанию BATCH — после обработки всей пачки из poll().
enable.auto.commitSpring Kafka выключает — коммитит сам контейнер- AckMode: RECORD, BATCH, MANUAL (Acknowledgment.acknowledge()), MANUAL_IMMEDIATE
- Исключение в методе → DefaultErrorHandler: повторы с отсрочкой, затем recoverer (DLT)
А ещё concurrency — число потребителей в контейнере, не больше числа партиций
Подробно
Устройство. Для каждого @KafkaListener фабрика (ConcurrentKafkaListenerContainerFactory) создаёт контейнер: concurrency потребителей (KafkaMessageListenerContainer), каждый в своём потоке вызывает poll(), передаёт записи вашему методу и управляет offset. Метод может принимать запись, значение с заголовками или список (batch listener).
Коммит. Spring Kafka выключает автокоммит клиента и коммитит сам:
BATCH(по умолчанию) — после того, как обработаны все записи из очередного poll();RECORD— после каждой записи (надёжнее, но медленнее);MANUAL/MANUAL_IMMEDIATE— когда вы вызвалиacknowledgment.acknowledge()(например, после асинхронной обработки).
Это at-least-once: если приложение упадёт после обработки, но до коммита, записи придут снова. Обработчик должен быть идемпотентным.
Ошибки. Исключение из метода перехватывает DefaultErrorHandler: делает несколько повторов с BackOff (по умолчанию 9 повторов без задержки — стоит настроить), потом передаёт запись recoverer'у, например DeadLetterPublishingRecoverer в топик .DLT, и идёт дальше. Ошибки десериализации — через ErrorHandlingDeserializer, иначе ядовитая запись зациклит потребителя.
Транзакции: с transactionManager у контейнера offset коммитится в транзакции Kafka вместе с отправленными сообщениями.
Что хотят услышать
- Контейнер, его потоки и вызов метода.
- AckMode и момент коммита.
- At-least-once и идемпотентность.
- DefaultErrorHandler и DLT.
Частые ошибки
- Не настроить ErrorHandlingDeserializer — бесконечный цикл на битой записи.
- Оставить повторы по умолчанию без отсрочки.
Проверьте себя
Один верный вариант. Подумайте, потом откройте ответ.
Показать варианты и ответ
- ✗Метод сам вызывает poll(), а offset коммитит Kafka автоматически раз в 5 секунд через enable.auto.commit
- ✗Offset коммитится до вызова метода
- ✗Каждый вызов метода открывает новое соединение с брокером
- ✓Контейнер в своих потоках вызывает poll() и метод и коммитит offset по AckMode
Что спросят дальше
- Что такое AckMode.MANUAL?
- Почему concurrency ограничена partitions?
Разбор темы
- Spring Kafka
KafkaTemplate, @KafkaListener, listener containers, offset ack, error handling, retries и DLT.
Связанные вопросы
- Как на стороне consumer-а добиться доставки at-least-once и почему при этом появляются дубли? Middle
- Чем commitSync() отличается от commitAsync() у Kafka consumer? Senior
- У топика 6 партиций. Что даст параметр concurrency = 3 у @KafkaListener и что произойдёт при concurrency = 10? Middle
- Как в Spring Kafka обычно строят retry и dead letter topic? Senior
Разбор собеседования с sotto
После собеседования sotto разбирает разговор: оценка, каждый вопрос — хорошо, частично или слабо, чего не хватило в ответе и над чем поработать к следующему разу. Первые 30 минут бесплатно.