Kafka批处理监听器中反序列化异常的重试策略与实现

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.SerializationUtils;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;

import java.util.List;
import java.util.Map;

@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()) // 使用自定义反序列化器
                )
        );
        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 void configure(Map<String, ?> configs, boolean isKey) {
            // No-op
        }

        @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("Simulating deserialization error for: " + s);
                throw new DeserializationException("Simulated deserialization failure for: " + s, bytes, false);
            }
            retries = 0; // 重置计数器
            System.out.println("Deserialized successfully: " + s);
            return s;
        }

        @Override
        public void close() {
            // No-op
        }
    }
}

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

import java.util.List;
import java.util.Objects;

@Component
public class StringListener {

    @KafkaListener(
            topics = {"string-test"},
            groupId = "test",
            batch = "true",
            containerFactory = "myContainerFactory"
    )
    public void listen(List<Message<String>> messages, Acknowledgment acknowledgment) {
        boolean hasDeserializationError = false;

        for (Message<String> message : messages) {
            String payload = message.getPayload();
            if (payload == null) {
                // 载荷为null，检查是否是反序列化异常
                byte[] exceptionHeader = (byte[]) message.getHeaders().get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER);
                if (exceptionHeader != null) {
                    DeserializationException deserializationException = 
                        ListenerUtils.byteArrayToDeserializationException(exceptionHeader);

                    if (deserializationException != null) {
                        System.err.println("Detected deserialization error for a message in batch: " + deserializationException.getMessage());
                        hasDeserializationError = true;
                        // 这里不直接抛出，而是标记，待循环结束后统一处理，确保检查完所有消息
                    }
                }
            } else {
                System.out.println("Processed message: " + payload);
            }
        }

        if (hasDeserializationError) {
            // 如果批次中存在反序列化错误，则重新抛出异常，触发整个批次的重试
            System.err.println("Batch contains deserialization errors. Re-throwing to trigger retry.");
            // 抛出任意RuntimeException即可，DefaultErrorHandler会根据配置进行重试
            throw new RuntimeException("Batch failed due to deserialization error(s).");
        }

        // 如果没有反序列化错误，或者所有错误都已处理且不需重试，则提交偏移量
        acknowledgment.acknowledge();
        System.out.println("Batch processed and acknowledged successfully.");
    }
}