Kafka consumer exception and offset commits : <p>I've been trying to do some POC work for Spring Kafka. Specifically, I wanted to experiment with what are the best practices in terms of dealing with errors while consuming messages within Kafka. </p>

<p>I am wondering if anyone is able to help with:</p>

<ol>
<li>Sharing best practices surrounding what Kafka consumers should do
when there is a failure</li>
<li>Help me understand how AckMode Record works, and how to prevent commits to the Kafka offset queue when an exception is thrown in the listener method. </li>
</ol>

<p>The code example for 2 is given below:</p>

<p>Given that AckMode is set to RECORD, which according to the <a href="http://docs.spring.io/spring-kafka/reference/htmlsingle/#_reference" rel="noreferrer">documentation</a>:</p>

<blockquote>
  <p>commit the offset when the listener returns after processing the
  record.</p>
</blockquote>

<p>I would have thought the the offset would not be incremented if the listener method threw an exception. However, this was not the case when I tested it using the code/config/command combination below. The offset still gets updated, and the next message continues to be processed.</p>

<p>My config:</p>

<pre><code>    private Map&lt;String, Object&gt; producerConfigs() {
    Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092");
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
}

   @Bean
ConcurrentKafkaListenerContainerFactory&lt;Integer, String&gt; kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory&lt;Integer, String&gt; factory =
            new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
    factory.setConsumerFactory(new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs()));
    factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);
    return factory;
}
</code></pre>

<p>My code:</p>

<pre><code>@Component
public class KafkaMessageListener{
    @KafkaListener(topicPartitions = {@TopicPartition( topic = "my-replicated-topic", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0", relativeToCurrent = "true"))})
    public void onReplicatedTopicMessage(ConsumerRecord&lt;Integer, String&gt; data) throws InterruptedException {
            throw new RuntimeException("Oops!");
    }
</code></pre>

<p>Command to verify offset:</p>

<pre><code>bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test-group
</code></pre>

<p>I'm using kafka_2.12-0.10.2.0 and org.springframework.kafka:spring-kafka:1.1.3.RELEASE</p>
