Spring-Cloud-Streams Kafka - How to stop the consumers

Question

I have a Spring-Cloud-Streams client reading from a Kakfa topic consisting of several partitions. The client calls a webservice for every Kafka message it reads. If the webservice is unavailable after a few retries, I want to stop the consumer from reading from Kafka. Referring to a previous Stackoverflow question (Spring cloud stream kafka pause/resume binders) I autowired BindingsEndpoint and call the changeState() method to try to stop the consumer but the logs show the consumer continuing to read the messages from Kafka after changeState() is invoked.

I am using Spring Boot version 2.1.2.RELEASE with Spring Cloud version Greenwich.RELEASE. The managed version for spring-cloud-stream-binder-kafka is 2.1.0.RELEASE. I have set the properties autoCommitOffset=true and autoCommitOnError=false.

Below is snippet of my codes. Is there something I have missed? Is the first input parameter to changeState() supposed to be the topic name?

If I want the consumer application to exit when the webservice is not available, can I simply do System.exit() without needing to stop the consumer first?

@Autowired
private BindingsEndpoint bindingsEndpoint;

...
...
@StreamListener(MyInterface.INPUT)  
    public void read(@Payload MyDTO dto,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
            @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {
        try {

            logger.info("Processing message "+dto);
            process(dto); // this is the method that calls the webservice


        } catch (Exception e) {

            if (e instanceof IllegalStateException || e instanceof ConnectException) {

                bindingsEndpoint.changeState("my.topic.name", 
                    BindingsEndpoint.State.STOPPED);    
                // Binding<?> b = bindingsEndpoint.queryState("my.topic.name"); ==> Using topic name returns a valid Binding object                     
            }

                e.printStackTrace();
                throw (e);
  }
}

Oleg Zhurakousky Oleg Zhurakousky · Accepted Answer · 2019-04-05T13:08:50

You can do so by utilising Binding visualization and control feature where you can visualize as well as stop/start/pause/resume bindings.

Also, you are aware that System.exit() will shut down the entire JVM?

Spring-Cloud-Streams Kafka - How to stop the consumers

2 Answers