Spring Kafka批处理监听器中反序列化错误的优雅重试机制

package com.example.kafka.config;

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.DeserializationException; // 导入 DeserializationException

@Configuration
@EnableKafka
public class KafkaConfig {

    @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())
                )
        );
        factory.getContainerProperties().setAckMode(
                ContainerProperties.AckMode.MANUAL_IMMEDIATE
        );

        // 配置 DefaultErrorHandler 以允许重试 DeserializationException
        DefaultErrorHandler errorHandler = new DefaultErrorHandler();
        // 关键步骤：从不可重试列表中移除 DeserializationException
        errorHandler.removeClassification(DeserializationException.class);
        // 可以选择配置退避策略，例如：
        // errorHandler.setBackOff(new FixedBackOff(1000L, 3L)); // 每次重试间隔1秒，共重试3次
        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; // 重置重试计数
            return s;
        }
    }
}

package com.example.kafka.listener;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.ListenerUtils;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.kafka.support.serializer.SerializationUtils;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.List;

@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) {
            // 检查消息负载是否为null，这通常意味着反序列化失败
            if (message.getPayload() == null) {
                // 尝试从消息头部获取原始的反序列化异常
                byte[] exceptionBytes = (byte[]) message.getHeaders().get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER);
                if (exceptionBytes != null) {
                    DeserializationException deserializationException =
                            ListenerUtils.byteArrayToDeserializationException(exceptionBytes);
                    System.err.println("检测到反序列化错误，消息内容可能为空。原始异常信息: " + deserializationException.getMessage());
                    // 重新抛出异常，以触发 DefaultErrorHandler 的重试机制
                    throw deserializationException;
                } else {
                    // 如果负载为null但头部没有异常信息，可能是其他原因，需要进一步调查
                    System.err.println("警告：消息负载为null，但未找到反序列化异常头部。");
                    // 也可以选择在这里抛出异常，或者记录并跳过
                }
            } else {
                // 消息反序列化成功，正常处理
                System.out.println("成功处理消息: " + message.getPayload());
            }
        }
        // 只有当所有消息都成功处理或重试机制已触发时才进行ack
        acknowledgment.acknowledge();
    }
}