title: Kafka Consumer: Stop processing messages when exception was raised
body: <p>I'm a bit confused about the poll() behaviour of (Spring) Kafka after/when stopping the <code>ConcurrentMessageListenerContainer</code>.</p>
  
  <p>What I want to achieve: 
  Stop the consumer after an exception was raised (for example message could not be saved to the database), do not commit offset, restart it after a given time and start processing again from the previously failed message.</p>
  
  <p>I read this article which says that the container will call the listener with the remaining records from the poll (<a href="https://github.com/spring-projects/spring-kafka/issues/451" rel="noreferrer">https://github.com/spring-projects/spring-kafka/issues/451</a>) which means that there is no guarantee that after the failed message a further message which was processed successfully will commit the offset. This could end up in lost/skipped messages.</p>
  
  <p>Is this really the case and if yes is there a solution to solve this without upgrading the newer versions? (DLQ is not a solution for my case)</p>
  
  <p>What I already did:
  Setting the <code>setErrorHandler()</code> and <code>setAckOnError(false)</code></p>
  
  <pre><code>private Map&lt;String, Object&gt; getConsumerProps(CustomKafkaProps kafkaProps,  Class keyDeserializer) {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      //Set common props
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProps.getBootstrapServers());
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
      props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProps.getConsumerGroupId());
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Start with the first message when a new consumer group (app) arrives at the topic
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // We will use "RECORD" AckMode in the Spring Listener Container
  
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
  
      if (kafkaProps.isSslEnabled()) {
          props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
          props.put("ssl.keystore.location", kafkaProps.getKafkaKeystoreLocation());
          props.put("ssl.keystore.password", kafkaProps.getKafkaKeystorePassword());
          props.put("ssl.key.password", kafkaProps.getKafkaKeyPassword());
      }
  
      return props;
  }
  </code></pre>
  
  <p>Consumer</p>
  
  <pre><code>public ConcurrentMessageListenerContainer&lt;String, byte[]&gt; kafkaReceiverContainer(CustomKafkaProps kafkaProps) throws Exception {
      StoppingErrorHandler stoppingErrorHandler = new StoppingErrorHandler();
  
      ContainerProperties containerProperties = new ContainerProperties(...);
      containerProperties.setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);
      containerProperties.setAckOnError(false);
      containerProperties.setErrorHandler(stoppingErrorHandler);
  
      ConcurrentMessageListenerContainer&lt;String, byte[]&gt; container = ...
      container.setConcurrency(1); //use only one container
      stoppingErrorHandler.setConcurrentMessageListenerContainer(container);
  
      return container;
  }
  </code></pre>
  
  <p>Error Handler</p>
  
  <pre><code>public class StoppingErrorHandler implements ErrorHandler {
  
      @Setter
      private ConcurrentMessageListenerContainer concurrentMessageListenerContainer;
  
      @Value("${backends.kafka.consumer.halt.timeout}")
      int consumerHaltTimeout;
  
      @Override
      public void handle(Exception thrownException, ConsumerRecord&lt;?, ?&gt; record) {
          if (concurrentMessageListenerContainer != null) {
              concurrentMessageListenerContainer.stop();
          }
  
          new Timer().schedule(new TimerTask() {
              @Override
              public void run() {
                  if (concurrentMessageListenerContainer != null &amp;&amp; !concurrentMessageListenerContainer.isRunning()) {
                      concurrentMessageListenerContainer.start();
                  }
              }
          }, consumerHaltTimeout);
      }
  }
  </code></pre>
  
  <p>What I'm using:</p>
  
  <pre><code>  &lt;groupId&gt;org.springframework.integration&lt;/groupId&gt;
    &lt;artifactId&gt;spring-integration-kafka&lt;/artifactId&gt;
    &lt;version&gt;2.1.2.RELEASE&lt;/version&gt;
  
    &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
    &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;
    &lt;version&gt;1.1.7.RELEASE&lt;/version&gt;
  </code></pre>