Блог

Пример использования: учитывая тему со 100 сообщениями в теме kafka, я хочу прочитать сообщение со смещением 10 до смещения 20. Я мог бы извлечь данные из начального смещения. когда я достигаю конечного смещения, я написал код для остановки контейнера.Даже после выполнения кода потребитель может использовать дополнительные сообщения (со смещения 21). Он останавливается только после чтения всех сообщений в теме

 @Service
public class Consumer1  implements MessageListener<String, GenericRecord> {

 @Override
  public void onMessage(ConsumerRecord<String, GenericRecord> data) {
    log.info("feed record {}", data);
    if (data.offset() == 20) {
      feedService.stopConsumer();
    }
  }
}

@Service
public class FeedService{

    // start logic here

   public void stopConsumer() {
    kafkaMessageListenerContainer.stop();
  }

}
 

Примечание: я использую последнюю версию spring-kafka (2.6.4). Одно из наблюдений заключается в том, что выполняется метод остановки контейнера, но потребитель не закрывается.И никаких ошибок на выходе

stop() Не завершает пакетный цикл текущих записей:

 while (isRunning()) {
            try {
                pollAndInvoke();
            }
            catch (@SuppressWarnings(UNUSED) WakeupException e) {
                // Ignore, we're stopping or applying immediate foreign acks
            }
 

Это pollAndInvoke() вызывает a KafkaConsumer.poll() , получает некоторую коллекцию записей и вызывает вашу onMessage() для каждой записи. В какой-то момент вы решаете вызвать остановку, но это не значит, что мы действительно находимся в конце этого списка записей, чтобы немедленно выйти.

Мы действительно останавливаемся на следующем цикле, когда это isRunning() уже возвращается false для нас.