title: @StreamListener (kafka) : How to stop consumers in-order to interrupt the processing
body: <p>Below is how we are publishing messages to topic,</p>
  
  <pre><code>for(int i=0;i&lt;=10000;i++){
  
    send("topic_name", "message");
  
  }
  </code></pre>
  
  <p>We have stream listener as below for above topic,</p>
  
  <pre><code>   @StreamListener("topic_name")
      public void process(KStream&lt;String, String&gt; inputKStream) {
          //processing of messages
      }
  </code></pre>
  
  <p>Now the requirement is we need to stop processing of messages when API is triggered.</p>
  
  <p>For that we have a controller as below, </p>
  
  <pre><code>@Autowired
  private CustomBindingEndpoint endpoint;
  
  @PutMapping(value = "stop", produces = APPLICATION_JSON_UTF8_VALUE)
  public void stopProcess() {
      endpoint.changeState("topic_name", State.STOPPED);
  }
  </code></pre>
  
  <p>The CustomBindingEndpoint.java is same as <a href="https://github.com/spring-cloud/spring-cloud-stream/issues/1650#issuecomment-476626008" rel="nofollow noreferrer">https://github.com/spring-cloud/spring-cloud-stream/issues/1650#issuecomment-476626008</a></p>
  
  <p>Once we trigger this API we expect the above StreamListener("topic_name") should get stop(further consuming has to be stopped)</p>
  
  <p>But once we trigger, even though binding.stop() is invoked for "topic_name" the consumer still consumes messages.</p>
  
  <p>I understand that binding.stop() will stop the consumer from consuming message. Please correct if i am wrong. kindly suggest how we can achieve this.</p>