Spring Cloud Stream批量模式下BatchMessagingMessageConverter可信类型配置问题

针对你使用Spring Boot 2.7.4 + Spring Cloud 2021.0.4时，Kafka批量消费模式下自定义Header被反序列为DefaultKafkaHeaderMapper$NonTrustedHeaderType的问题，以下是可行的解决方式：

核心原因 #

批量消费模式下，Spring Cloud Stream使用BatchMessagingMessageConverter处理消息，它的HeaderMapper默认未包含自定义类所在包的信任列表，常规的spring.kafka配置无法直接作用于该转换器的Header反序列化逻辑。

方案一：自定义批量消息转换器并配置可信包 #

通过自定义BatchMessagingMessageConverter，手动为其HeaderMapper添加可信包，确保批量消费时能正确反序列化自定义Header类：

@Configuration
public class KafkaBatchHeaderConfig {

    @Bean
    public BatchMessagingMessageConverter batchMessagingMessageConverter(KafkaHeaderMapper headerMapper) {
        BatchMessagingMessageConverter converter = new BatchMessagingMessageConverter();
        converter.setHeaderMapper(headerMapper);
        return converter;
    }

    @Bean
    public KafkaHeaderMapper kafkaHeaderMapper() {
        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
        // 添加自定义Header类所在的包路径
        mapper.addTrustedPackages("com.yourproject.custom.headers");
        // 也可直接指定具体类的包名
        // mapper.addTrustedPackages(CustomHeaderClass.class.getPackageName());
        return mapper;
    }
}

该配置无需额外指定headerMapperBeanName，Spring Cloud Stream会自动在批量消费模式下使用自定义的转换器。

方案二：绑定级配置直接指定可信包 #

针对目标输入绑定，通过配置直接设置HeaderMapper的可信包，适配Spring Cloud Stream Kafka 2021.0.x版本：

# 替换为你的输入绑定名称，此处对应nameBind-in-0
spring.cloud.stream.kafka.bindings.nameBind-in-0.consumer.header-mapper-trusted-packages=com.yourproject.custom.headers

该配置直接作用于批量消费的HeaderMapper，无需编写自定义Bean。

为什么之前的配置无效？ #

• spring.kafka.consumer.properties.spring.json.trusted.packages：此配置仅针对Kafka消息体的JSON反序列化，不作用于Header的反序列化逻辑。
• spring.cloud.stream.kafka.binder.headerMapperBeanName：批量模式下，BatchMessagingMessageConverter的HeaderMapper优先级更高，单独指定HeaderMapper Bean无法自动关联到批量转换器，因此无效。

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