如何在消费者停止后停止 Spring Kafka Listener 并调用相同的函数来执行？

Question

我有一个场景，我需要停止 kafka 并在 Consumer 停止后调用一些函数来执行。同样，流程是这样的：

1. 使用来自 kafka 主题的消息
2. 将每个消费的消息添加到文件中
3. 如果过去 10 秒内没有收到任何消息，则停止 kafka 侦听器
4. 调用一些函数例如：UploadFileToS3()

我在我的消费者方法上使用 spring kafka 的 @KafkaListener 注释。

我知道停止使用 @KafkaListener 注释的 Kafka 消费者我可以使用 KafkaListenerEndpointRegistry

我想要这样的东西：

if(not consumed messages for more than 10s) {
    kafkaListenerEndpointRegistry.getListenerContainer("my-listener-id").stop();
    UploadFileToS3()
}

如何使用 Spring Kafka 做到这一点？

Answer 1

见ListenerContainerIdleEvent:

/**
 * An event that is emitted when a container is idle if the container
 * is configured to do so.
 *
 * @author Gary Russell
 *
 */
public class ListenerContainerIdleEvent extends KafkaEvent {

文档在这里：https://docs.spring.io/spring-kafka/docs/current/reference/html/#idle-containers