在读取所有消息直到某个特定时间之后，停止KafkaListener(Spring Kafka Consumer) [英] Stop KafkaListener ( Spring Kafka Consumer) after it has read all messages till some specific time

问题描述

我正在尝试从一个分区主题安排我的消费过程.我可以使用 endpointlistenerregistry.start()启动它，但是我想在消耗完当前分区中的所有消息后即当到达当前分区中的最后一个偏移量时停止它.我完成消费并关闭后，就可以完成对该主题的制作.在启动调度程序并停止消费者之前，如何确保我已阅读所有消息?我正在为消费者使用 @Kafkalistener .

I am trying to schedule my consumption process from a single partition topic. I can start it using endpointlistenerregistry.start() but I want to stop it after I have consumed all the messages in current partition i.e. when I reach to last offset in current partition. Production into the topic is done after I have finished the consumption and close it. How should I achieve the assurance that I have read all the messages till the time I started scheduler and stop my consumer ? I am using @Kafkalistener for consumer.

推荐答案

设置 idleEventInterval 容器属性，并添加 @EventListener 方法以侦听 ListenerContainerIdleEvent s.

Set the idleEventInterval container property and add an @EventListener method to listen for ListenerContainerIdleEvents.

然后停止容器.

这篇关于在读取所有消息直到某个特定时间之后，停止KafkaListener(Spring Kafka Consumer)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！