apache-kafka - 如何优雅地关闭正在运行的 Kafka Consumer

Question

我需要根据一些数据库驱动的属性来打开/关闭 Kafka 消费者。怎么可能实现。

我想到的一种方法是:当消费者标志关闭时，从消费者那里抛出异常。容器工厂配置定义为

factory.setErrorHandler(new SeekToCurrentErrorHandler());

但它会主动寻找相同的信息。

有什么方法可以关闭心跳，然后按需重新开启。

Answer 1

你可以stop()和start()监听容器。

您似乎在使用 @KafkaListener，因为您使用的是容器工厂。

这样的话

@KafkaListener(id = "foo" ...)

然后使用 KafkaListenerEndpointRegistry bean ...

registry.getListenerContainer("foo").stop();