Как я могу повторить сообщения об ошибках от kafka?

Question

Мои spring-boot application(consumer) обрабатывают сообщения от Apache Kafka.Периодически, массаж не может обработать, и потребитель бросает исключение.Потребитель в любом случае совершает смещение.Можно ли отличить сообщения об успехах от сообщений об ошибках в Kafka?Я думаю, я не могу.Это правда?Если это правда, у меня есть главный вопрос:

Как я могу повторить сообщения об ошибках ?Я знаю некоторые способы, но я не уверен в их правильности.

1) Изменить смещение на раннее.Но таким образом сообщения об успехе будут повторяться тоже.

2) Когда я ловлю исключение, я отправляю это сообщение в другую тему (например, error-topic).Но это выглядит сложно.

3) Что-то еще (ваш вариант)

Answer 1

Если вы хотите хотя бы один раз гарантировать, общая схема выглядит следующим образом:

• Отключить автоматическую фиксацию (установите enable.auto.commit в false)
• Использование сообщений

• Для каждого сообщения:

  • Если ошибок нет, то смещение коммита
  • В случае ошибки повторите столько раз, сколько пожелаете
  • В случае успеха, совершить
  • Если вы хотите отказаться, войти или опубликовать в очереди ошибок (для анализа или более поздней попытки)

• Повторите

Answer 2

Использовать SeekToCurrentErrorHandler . Будет смещено смещение для воспроизведения сообщения (по умолчанию 10 раз, но настраивается).

После того, как повторные попытки исчерпаны, он вызывает «восстановитель», который может предпринять какое-либо действие, например DeadLetterPublishingRecoverer .