Kafka批处理监听器中反序列化错误重试策略详解

@org.springframework.context.annotation.Configuration
@EnableKafka
public class KafkaConfiguration {

    @Bean("myContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> createFactory(
            KafkaProperties properties
    ) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(
                new DefaultKafkaConsumerFactory<>(
                        properties.buildConsumerProperties(),
                        new StringDeserializer(),
                        new ErrorHandlingDeserializer<>(new MyDeserializer()) // 使用ErrorHandlingDeserializer包装自定义反序列化器
                )
        );
        factory.getContainerProperties().setAckMode(
                ContainerProperties.AckMode.MANUAL_IMMEDIATE
        );

        // 创建并配置DefaultErrorHandler
        DefaultErrorHandler errorHandler = new DefaultErrorHandler();
        // 允许对DeserializationException进行重试
        errorHandler.removeClassification(DeserializationException.class); 
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }

    // 示例自定义反序列化器，模拟间歇性错误
    static class MyDeserializer implements Deserializer<String> {
        private int retries = 0;

        @Override
        public String deserialize(String topic, byte[] bytes) {
            String s = new String(bytes);
            // 模拟在第一次尝试时遇到特定消息（包含"7"）时抛出异常
            if (s.contains("7") && retries == 0) {
                retries = 1; // 标记已尝试一次
                System.out.println("模拟反序列化失败: " + s);
                throw new RuntimeException("模拟反序列化错误");
            }
            retries = 0; // 重置计数器
            System.out.println("成功反序列化: " + s);
            return s;
        }
    }
}

@Component
public class StringListener {

    @KafkaListener(
            topics = {"string-test"},
            groupId = "test",
            batch = "true",
            containerFactory = "myContainerFactory"
    )
    public void listen(List<Message<String>> messages, Acknowledgment acknowledgment) {
        for (Message<String> message : messages) {
            String payload = message.getPayload();
            if (payload == null) {
                // 检查消息头中是否存在反序列化异常
                byte[] exceptionHeader = message.getHeaders().get(
                        SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER, byte[].class
                );
                if (exceptionHeader != null) {
                    DeserializationException deserializationException = 
                        ListenerUtils.byteArrayToDeserializationException(exceptionHeader);
                    // 打印异常信息，然后重新抛出，触发批次重试
                    System.err.println("检测到反序列化异常，将重试批次: " + deserializationException.getMessage());
                    throw deserializationException; // 重新抛出异常，整个批次将被重试
                }
            }
            System.out.println("处理消息: " + payload);
        }
        acknowledgment.acknowledge(); // 批次所有消息处理成功后手动提交偏移量
    }
}