Как изящно выключить работающий Kafka Consumer

Question

Мне нужно включить / выключить потребителя Kafka на основе некоторого свойства, управляемого базой данных.Как это может быть достигнуто.

Один из способов, о котором я подумал, это: исключение из потребителя, когда флаг потребителя выключен.и конфигурация фабрики контейнеров определена как

factory.setErrorHandler(new SeekToCurrentErrorHandler());

Но она активно ищет то же сообщение.

есть ли способ отключить сердцебиение и снова включить его по требованию.

Answer 1

Вы можете stop() и start() контейнер слушателя.

Похоже, вы используете @KafkaListener, поскольку вы используете фабрику контейнеров.

В этом случае

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

, а затем используйте KafkaListenerEndpointRegistry bean ...

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

Answer 2

Потребитель имеет несколько API для управления своим состоянием:

• pause() / resume(): позволяет остановить /возобновить потребление из набора разделов.Потребитель остается подписанным (таким образом, нет перебалансировки), но просто не получает новые записи, пока не будет возобновлено

• unsubscribe(): Позволяет изменить подписку потребителя, если не подписанк чему-либо, он просто останется подключенным к кластеру.

Если вы «покончили» с потребителем, вы также можете позвонить close() и начатьновый при необходимости