Consume messages from Kafka only if some condition true

Question

We have specific topic and it is required to consume messages only if condition consumeEnabled=true. So, it should work like this:

1. If application is starting up and consumeEnabled=true, then assign partitions to consumers and consume message from topic.
2. If application is starting up and consumeEnabled=false, then don't assign partitions to consumers and don't consume message from topic.
3. If application already run with consumeEnabled=false, but in runtime property became consumeEnabled=true, then in runtime assign partitions to consumers and consume message from topic.

Case when app is consuming messages, but then consumeEnabled become false not necessary to consider.

Please, hep define the best way to implement decision with Spring Kafka and\or Kafka Java client

Answer 1

If you are using @KafkaListener then

@KafkaListener(id = "foo", ... , autoStartup="${consume.enabled}")

where consume.enabled is a property.

To start/stop a container at runtime, use the KafkaListenerEndpointRegistry bean.

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

Answer 2

You could put your consumer in a simple thread that toggles the polling state of the consumer object.

public class EnabledConsumer implements Runnable {

    private Consumer consumer;
    private boolean enabled;

    public EnabledConsumer(Consumer consumer, boolean enabled) {
        this.consumer = consumer;
        this.enabled = enabled;
    }

    public void setEnabled(boolean enable) {
        this.enabled = enable;
    }

    @Override
    public void run() {
        while(enabled) {
            ConsumerRecords records = consumer.poll(...);
            ...
        }

}