title: How to achieve Always retry policy/Custom Retry Policy for a Kafka Consumer without causing a rebalance in the group
body: <p>I am trying to write a resilient Kafka consumer. If there occurs an exception while processing the message in the listener method, I would like to retry it. I would like to retry a few times for certain exceptions, always for certain exceptions and never for the rest. I have read the Spring docs regarding SeekToCurrentErrorHandler, but not 100% sure how to implement it.</p>
  <p>I have subclassed the ExceptionClassifierRetryPolicy and and am returning the appropriate Retry Policy based on the exceptions that occur in the Listener method.</p>
  <p>I have created RetryTemplate and have and set its RetryPolicy with my custom implementation in the subclass.</p>
  <p>I have set the retryTemplate on the Kafka container. I have set the error handler to be new SeekToCurrentHandler and also set the stateful retry property to be true.</p>
  <p>Listener Method</p>
  <pre><code>@KafkaListener(topics = &quot;topicName&quot;, containerFactory = &quot;containerFactory&quot;)
     public void listenToKafkaTopic(@Payload Message&lt;SomeAvroGeneratedClass&gt; message, Acknowledgement ack){
        SomeAvroGeneratedClass obj = message.getPayLoad();
        processIncomingMessage(obj);
        ack.acknowledge();
     }
  </code></pre>
  <p>Custom Retry Policy class</p>
  <pre><code> @Component
   public class MyRetryPolicy extends ExceptionClassifierRetryPolicy
     {
        @PostConstruct
         public void init(){
               final SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
               simpleRetryPolicy.setMaxAttempts(8);
  
       this.setExceptionClassifier( classifiable -&gt;
             {
          // Always Retry when instanceOf   TransientDataAccessException
         if( classifiable.getCause() instanceof TransientDataAccessException)
              {
                 return new AlwaysRetryPolicy();                                         
  
              }
        else if(classifiable.getCause() instanceOf NonTransientDataAccessException)
             {
                return new NeverRetryPolicy();
             }
            else
              {
              return simpleRetryPolicy;
              }
  
        } );
   }}
  </code></pre>
  <p>Retry Template and Container Config</p>
  <pre><code>@Configuration
  public class RetryConfig{
  
  
   @Bean
   public RetryTemplate retryTemplate(@Autowired ConcurrentKafkaListenerContainerFactory factory){
  
     RetryTemplate retryTemplate = new RetryTemplate();
     retryTemplate.setRetryPolicy(new MyRetryPolicy());
     FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy()
     fixedBackOffPolicy.setBackOffPeriod(2000l);
     retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
  
     factory.setRetryTemplate(retryTemplate);
     factory.setAckOnError(false);
     factory.setErrorHandler(new SeekToCurrentErrorHandler());
     factory.setStateFulRetry(true);
     factory.setRecoveryCallback(//configure recovery after retries are exhausted and commit offset
     );
  
    }
  }
  </code></pre>
  <p>Listener properties:</p>
  <ol>
  <li>AckMode = Manual</li>
  <li>auto.offset.commit = false</li>
  </ol>
  <p>Question:</p>
  <ol>
  <li><p>With my current code, can I achieve my retry logic defined in MyRetryPolicy without causing a consumer rebalance when AlwaysRetryPolicy is returned? If no, please direct me in the right path.</p>
  </li>
  <li><p>Is my approach right in using the Error handler as well as the Retry?</p>
  </li>
  </ol>