如何为Kafka消费者实现始终重试策略/自定义重试策略，而不会导致组中的重新平衡

Question

我试图写一个有弹性的Kafka消费者。如果在侦听器方法中处理消息时发生异常，我想重试。我想对某些异常重试几次，对于某些异常，从不对某些异常重试。我已经阅读了有关“搜索到当前错误处理者”的Spring文档，但不是100%确定如何实现它。

我已经子类化了ExceptionClatefierRetryPolicy，并根据Listener方法中发生的异常返回适当的重试策略。

我已经创建了RetryTemplate，并在子类中使用我的自定义实现设置了它的RetryPolicy。

我已在Kafka容器上设置了retryTemplate。我已将错误处理程序设置为new SeekToCurrentHandler，并将有状态重试属性设置为true。

侦听器方法

@KafkaListener(topics = "topicName", containerFactory = "containerFactory")
   public void listenToKafkaTopic(@Payload Message<SomeAvroGeneratedClass> message, Acknowledgement ack){
      SomeAvroGeneratedClass obj = message.getPayLoad();
      processIncomingMessage(obj);
      ack.acknowledge();
   }

自定义重试策略类

 @Component
 public class MyRetryPolicy extends ExceptionClassifierRetryPolicy
   {
      @PostConstruct
       public void init(){
             final SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
             simpleRetryPolicy.setMaxAttempts(8);

     this.setExceptionClassifier( classifiable ->
           {
        // 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;
            }

      } );
 }}

重试模板和容器配置

@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
   );

  }
}

侦听器属性：

1. AckMode=手动
2. auto.offset.commit=false

问题：

> < li>

使用我当前的代码，当AlwaysRetryPolicy返回时，我能否实现MyRetryPolicy中定义的重试逻辑，而不导致使用者重新平衡？如果没有，请指引我正确的道路。

我的方法在使用错误处理程序和重试方面是否正确？

Answer 1

有状态重试与＜code＞SeekToCurrentErrorHandler＜/code＞是正确的方法。

但是，如果您使用的是最新版本（2.2或更高版本），则错误处理程序将在10次尝试后放弃;您可以通过将最大故障设置为-1（2.2）或回退以长.MAX_VALUE（2.3或更高）来使无穷大。