Ручное подтверждение сообщения в Kafka

В этой статье описывается реализация задачи ручного подтверждения(acknowledgment) обработки сообщения в Kafka через ручную отправку смещения(commit offset) сообщения. Логика реализована с использованием Java, Spring и Kafka.

План:

1. Реализовать слушатель Kafka с ручным подтверждением.

2. Добавить файл application.properties с инструкциями для ручной отправки смещения.

3. Описание механизма подтверждения в Kafka и определения.

1. Реализовать слушатель Kafka с ручным подтверждением

Создадим слушатель Kafka через использование аннотации @KafkaListener. Далее добавим переменную Acknowledgment для дальнейшей настройки подтверждения:

@Service public class ServiceCallKafkaListener {      @KafkaListener(id = "listenerId",             groupId = "groupListenerId",             topics = "topicName")     public void listenServiceCall(@Payload String message,                                   Acknowledgment acknowledgment) {         //here is your logic for message processing         boolean logicForMessageProcessingCompleted = true;         if (logicForMessageProcessingCompleted) {             //manual commit             acknowledgment.acknowledge();         }     } }

Нужно вызвать метод acknowledge() объекта Acknowledgment в теле метода слушателя. Метод слушателя это: listenServiceCall(). Метод acknowledge() отправляет сигнал в Kafka о том, что он подтверждает обработку сообщения.

2. Добавим файл application.properties и настройки

Добавим настройки Kafka в файл application.properties для того, что бы механизм ручной отправки смещения начал работать.

#for detail description see "Resources" section in this article #type of acknowledgment mode spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE  #property that turns off auto commit spring.kafka.consumer.enable-auto-commit=false

3. Описание механизма подтверждения Kafka и определения

Подтверждение(Acknowledgment) — это сигнал в Kafka, который означает, что сообщение обработано слушателем и слушатель готов слушать следующее сообщение. Слушателем в этой статье является метод listenServiceCall. Kafka двигает указатель смещения до следующего, после отправки сигнала о подтверждении(acknowledge()).

Cлушатель(KafkaListener) будет получать одно и то же сообщение с одним и тем же смещением(offset), до тех пор пока слушатель не отправит сигнал о подтверждении.

Смещение(offset) в Kafka — это идентификатор сообщения в очереди сообщений.

Источники для дополнительного чтения

1. Acknowledge mode description

2. Acknowledge method documentation