如何重试批量监听器遇到的Kafka反序列化错误？

针对你遇到的Avro Schema主机连接导致的间歇性反序列化错误，无需复杂自定义failedDeserializationFunction，通过结合ErrorHandlingDeserializer和DefaultErrorHandler的退避策略即可实现可配置的重试机制，具体步骤如下：

关键配置修改 #

1. 让反序列化失败时抛出异常
   默认ErrorHandlingDeserializer在失败时返回null，这会导致消息被直接传递到监听器而非触发重试。我们需要配置它在失败时抛出异常，将错误传递给ErrorHandler。

2. 配置退避重试策略
   通过DefaultErrorHandler搭配FixedBackOff或ExponentialBackOff，指定重试间隔和最大重试次数。

修改后的完整配置类 #

@Configuration
@EnableKafka
public class Configuration {
    @Bean("myContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> createFactory(
            KafkaProperties properties
    ) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        
        // 配置ErrorHandlingDeserializer，反序列化失败时抛出异常
        var errorHandlingDeserializer = new ErrorHandlingDeserializer<>(new MyDeserializer());
        errorHandlingDeserializer.setFailedDeserializationFunction((topic, data, exception) -> {
            throw new KafkaException("Deserialization failed for topic: " + topic, exception);
        });
        
        factory.setConsumerFactory(
                new DefaultKafkaConsumerFactory(
                        properties.buildConsumerProperties(),
                        new StringDeserializer(),
                        errorHandlingDeserializer
                )
        );
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        // 配置退避重试：每次间隔1秒，最多重试3次
        factory.setCommonErrorHandler(new DefaultErrorHandler(
                new FixedBackOff(1000L, 3L)
        ));
        
        return factory;
    }

    // 模拟间歇性反序列化失败的测试反序列化器
    static class MyDeserializer implements Deserializer<String> {
        int retries = 0;
        @Override
        public String deserialize(String topic, byte[] bytes) {
            String s = new String(bytes);
            if (s.contains("7") && retries == 0) {
                retries = 1;
                throw new RuntimeException("Simulated schema host connection failure");
            }
            retries = 0;
            return s;
        }
    }
}

配置说明 #

• ErrorHandlingDeserializer的异常抛出：通过setFailedDeserializationFunction自定义失败逻辑，直接抛出KafkaException，确保错误能被DefaultErrorHandler捕获并触发重试。
• 退避策略：示例中使用FixedBackOff(1000L, 3L)表示每次重试间隔1秒，最多重试3次。如果需要指数退避，可替换为ExponentialBackOff，比如：

  new ExponentialBackOff(1000L, 2.0) // 初始间隔1秒，每次间隔翻倍
      .setMaxAttempts(5) // 最多重试5次
  

• 批量消费适配：你当前使用批量消费模式，DefaultErrorHandler默认会重试整个批次，符合反序列化错误的场景（通常批次内消息依赖同一个Schema服务）。

额外注意事项 #

• 确保你的Spring Kafka版本支持上述配置（建议使用2.8+版本，DefaultErrorHandler在该版本后成为默认错误处理器）。
• 如果需要对特定异常类型进行重试控制，可以通过DefaultErrorHandler的addRetryableExceptions或addNotRetryableExceptions方法过滤异常。

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