title: Pause and Resume KafkaConsumer
body: <p>What I've to do is pause the <code>KafkaConsumer</code> if during message consuming an error is thrown.</p>
  <p>This is what I wrote</p>
  <pre><code>@KafkaListener(...)
  public void consume(
   @Header(KafkaHeaders.CONSUMER) KafkaConsumer&lt;String,String&gt; consumer,
   @Payload String message) {
    
      try {
          //consumer message
      } catch(Exception e) {
          saveConsumer(consumer);
          consumer.pause();
      }
  }
  </code></pre>
  <p>Then I wrote a REST service in order to resume the consumer</p>
  <pre><code>@RestController
  @RequestMapping(&quot;/consumer&quot;)
  class ConsumerRestController {
      @PostMapping(&quot;/resume&quot;)
      public void resume() {
          KafkaConsumer&lt;String,String&gt; consumer = getConsumer();
          if(consumer != null) {
              consumer.resume(consumer.paused());
          }
      }
  }
  </code></pre>
  <p>Now, I've two questions.
  First question: When I call consumer.pause() from <code>@KafkaListener</code> annotated method what happens?
  Consumer is immediately paused or I can receive other messages associated on other offset of same topic-partition.
  For example, I have &quot;message1&quot; with offset 3 and &quot;message2&quot; with offset 4, &quot;message1&quot; cause an exception, what happens to &quot;message2&quot;? Is it consumed anyway?</p>
  <p>Second question: Resuming the consumer from REST service give a <code>ConcurrentModificationException</code> because <code>KafkaConsumer</code> is not thread safe. So, how come I have to do this?</p>