title: Kafka retry mechanism doesn't stop even the previous retry attempt was successful
body: <p>I have a kafka retry mechanism in place which retry 2 times by waiting 30 seconds for each attempt. I noticed that even though the first retry attempt was successful, It's still retrying the second attempt. This results in generating duplicate messages in the kafka topic. Is there any way to stop Kafka to do unnecessary retries when the previous retry attempt is successful?</p>
  <p>Here is my listener configuration</p>
  <pre><code>  @Bean
    @ConditionalOnMissingBean(name = &quot;kafkaListenerContainerFactory&quot;)
    public ConcurrentKafkaListenerContainerFactory&lt;String, SpecificRecord&gt;
      kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, SpecificRecord&gt; factory =
        new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(testConsumerFactory());
      factory.getContainerProperties().setAckOnError(false);
      factory.getContainerProperties().setAckMode(AckMode.RECORD);
      SeekToCurrentErrorHandler errorHandler =
        new SeekToCurrentErrorHandler((record, exception) -&gt; {
          LOGGER.error(&quot;Error while processing the record {}&quot;, exception.getCause().getMessage());
        }, new FixedBackOff(30000L, 2L));
      factory.setErrorHandler(errorHandler);
      return factory;
    }
  </code></pre>
  <p>Here is my listener logic and the flow</p>
  <ol>
  <li>ConsumerA consumes the data from the topicA and  makes a call to microservice for the data</li>
  <li>After getting  the data, Producer publishes the data to topicB</li>
  <li>ConsumerB consumes the data from the topicB and makes a call to a microservice for persisting the data</li>
  <li>Once data gets persisted, a new message gets published to topicB.</li>
  </ol>
  <p>Consumer logic for topicA</p>
  <pre><code>    @KafkaListener(topics = &quot;${test.topicA.name}&quot;,
      containerFactory = &quot;kafkaListenerContainerFactory&quot;)
    public void topicListener(ConsumerRecord&lt;String, SpecificRecord&gt; record) {
      LOGGER.info(&quot;Consumed {} topic from partition {} &quot;, record.topic(), record.partition());
      testService.getData(record);
    }
  </code></pre>
  <p>Consumer logic for topicB</p>
  <pre><code> @KafkaListener(topics = &quot;${test.topicB.name}&quot;,
      containerFactory = &quot;kafkaListenerContainerFactory&quot;)
    public void topicListener(ConsumerRecord&lt;String, SpecificRecord&gt; record) {
      LOGGER.info(&quot;Consumed {} topic from partition {} &quot;, record.topic(), record.partition());
       testService2.persistDetails(record);
    }
  </code></pre>