如何在不引起组重新平衡的情况下为 Kafka 消费者实现始终重试策略/自定义重试策略

Question

我正在尝试编写一个有弹性的 Kafka 消费者。如果在侦听器方法中处理消息时出现异常，我想重试它。对于某些异常，我想重试几次，总是针对某些异常，从不针对其他异常。我已阅读有关 SeekToCurrentErrorHandler 的 Spring 文档，但不能 100% 确定如何实现它。

我对 ExceptionClassifierRetryPolicy 进行了子类化，并且正在根据 Listener 方法中发生的异常返回适当的重试策略。

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

我已经在 Kafka 容器上设置了 retryTemplate。我已将错误处理程序设置为新的 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

问题：

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

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

Answer 1

有状态重试与SeekToCurrentErrorHandler 结合使用是正确的方法。

但是，如果您使用的是最新版本（2.2 或更高版本），错误处理程序将在 10 次尝试后放弃；您可以通过将 maxFailures 设置为 -1 (2.2) 或将 backOff 设置为 Long.MAX_VALUE（2.3 或更高版本）来实现无穷大。