Spring Kafka 2.x至3.0.1迁移：错误处理器类改造疑问

核心问题解答 #

1. DefaultErrorHandler 中使用 handleRemaining 完全可行
   Spring Kafka 3.x 里，原 SeekToCurrentErrorHandler 已被 DefaultErrorHandler 替代，原 handle 方法的职责被 handleRemaining 承接——两者都是处理当前批次中剩余的未消费记录，所以改用 handleRemaining 是正确的替代方案。

2. SeekUtils.seekOrRecover 参数适配
   3.x 版本对 SeekUtils.seekOrRecover 的参数做了调整：

• 原带参数的 getSkipPredicate(records, thrownException) 方法被移除，现在直接通过 DefaultErrorHandler 的无参 getSkipPredicate() 获取全局跳过断言（类型为 BiPredicate<ConsumerRecord<?, ?>, Exception>）
• 新增 RecoveryStrategy 参数，可通过 DefaultErrorHandler 的 getRecoveryStrategy() 获取

迁移后的代码实现 #

@Component
@Slf4j
public class DeserializationFailedErrorHandler extends DefaultErrorHandler {   

    @Override
    public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
        SeekUtils.seekOrRecover(thrownException, 
                                records, 
                                consumer, 
                                container, 
                                isCommitRecovered(), 
                                getSkipPredicate(), 
                                logger, 
                                getLogLevel(),
                                getRecoveryStrategy());
        // 保留你的自定义业务逻辑
        // do somethings
    }
}

额外优化提示 #

• 若需自定义跳过逻辑，无需重写 getSkipPredicate，建议在构造函数中直接注入自定义断言：

  public DeserializationFailedErrorHandler() {
      super((record, ex) -> {
          // 示例：仅跳过反序列化异常
          return ex instanceof DeserializationException;
      });
  }
  

• DefaultErrorHandler 内置了更灵活的重试/恢复配置，比如通过 addNotRetryableExceptions() 指定无需重试的异常类型，可减少自定义 handleRemaining 中的冗余逻辑。

内容的提问来源于stack exchange，提问作者Bartek