在某个特定时间之前阅读完所有消息后，停止KafkaListener（Spring Kafka Consumer）

Question

我试图从单个分区主题安排我的消费过程。我可以使用endpointlistenerregistry启动它。start（）但我想在消耗完当前分区中的所有消息后，即当我到达当前分区中的最后一个偏移量时停止它。制作成主题是在我完成消费并关闭它之后完成的。我应该如何确保在启动scheduler并停止我的消费者之前，我已经阅读了所有消息？我正在为消费者使用Kafkalistener。

Answer 1

>

要读取到最后一个偏移量，您只需轮询直到获得空记录。

您可以调用KafkanConsumer。消费结束时暂停（）。在下一个计划中，需要调用KafkanConsumer。恢复（）。

暂停从请求的分区提取。在使用resume（Collection）恢复这些分区之前，将来对轮询（Duration）的调用不会返回这些分区中的任何记录。请注意，此方法不影响分区订阅。特别是，使用自动分配时，不会导致组重新平衡。

像这样的东西，

List<TopicPartition> topicPartitions = new ArrayList<>();
void scheduleProcess() {
    topicPartitions = ... // assign partition info for this
    kafkaConsumer.resume(topicPartitions)
    while(true) {
       ConsumerRecords<String, Object> events = kafkaConsumer.poll(Duration.ofMillis(1000));
       if(!events.isEmpty()) {
          // processing logic
       } else {
          kafkaConsumer.pause(List.of(topicPartition));
          break;
       }
    }
}

Answer 2

设置idleEventInterval容器属性，并添加一个EventListener方法来侦听ListenerContainerIdleEvent。

然后停止容器。