Spring Boot Kafka：如何让JsonDeserializer忽略类型头？

问题描述 #

Spring Kafka生产者会在消息中嵌入类型头，指定消费者应将消息反序列化为对应类。当生产者未使用Spring Kafka但消费者使用时，会出现问题：JsonDeserializer无法反序列化消息，抛出异常No type information in headers and no default type provided。
尝试过两种方案均无效：

• 设置默认反序列化类型，但单个主题含多种消息模式时该方法不适用；
• 在application.properties中设置spring.kafka.consumer.properties.spring.json.use.type.headers=false，但操作后仍抛出相同异常。

有效解决方法 #

1. 直接在自定义JsonDeserializer实例中配置 #

不要通过消费者通用属性间接配置，直接在创建JsonDeserializer实例时设置useTypeHeaders=false，确保配置优先级：

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka服务地址");
    // 其他基础消费者配置（如group.id等）
    
    JsonDeserializer<Object> deserializer = new JsonDeserializer<>(Object.class);
    deserializer.setUseTypeHeaders(false);
    // 添加信任包，避免反序列化安全限制
    deserializer.addTrustedPackages("com.your.company.*");
    return new DefaultKafkaConsumerFactory<>(configProps, new StringDeserializer(), deserializer);
}

2. 多消息类型场景：自定义类型映射策略 #

如果单个主题包含多种消息结构，可通过自定义TypeMapper完全绕过Spring Kafka默认的类型头逻辑，根据业务规则指定反序列化类型：

@Bean
public ConsumerFactory<String, Object> multiTypeConsumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka服务地址");
    // 其他基础消费者配置
    
    JsonDeserializer<Object> deserializer = new JsonDeserializer<>(Object.class);
    deserializer.setUseTypeHeaders(false);
    deserializer.addTrustedPackages("com.your.company.*");
    
    // 自定义类型映射逻辑，示例：从自定义消息头获取类型标识
    deserializer.setTypeMapper(new TypeMapper() {
        @Override
        public void fromHeaders(Headers headers, JavaType type) {}
        
        @Override
        public JavaType toJavaType(Headers headers) {
            // 根据实际业务场景调整，比如读取自定义头、解析消息字段判断类型
            Header msgTypeHeader = headers.lastHeader("message-type");
            if (msgTypeHeader != null) {
                String msgType = new String(msgTypeHeader.value());
                switch (msgType) {
                    case "user":
                        return TypeFactory.defaultInstance().constructType(User.class);
                    case "order":
                        return TypeFactory.defaultInstance().constructType(Order.class);
                    default:
                        return TypeFactory.defaultInstance().constructType(Object.class);
                }
            }
            return TypeFactory.defaultInstance().constructType(Object.class);
        }
    });
    
    return new DefaultKafkaConsumerFactory<>(configProps, new StringDeserializer(), deserializer);
}

3. 配置文件层面的正确写法（确保生效） #

如果偏好配置文件方式，需明确指定值反序列化器为JsonDeserializer，并直接配置其专属属性：

# 指定值反序列化器为JsonDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
# 禁用类型头依赖
spring.kafka.consumer.properties.spring.json.use.type.headers=false
# 必须配置信任包，否则会触发安全校验异常
spring.kafka.consumer.properties.spring.json.trusted.packages=com.your.company.*

注意：如果项目中存在自定义ConsumerFactory Bean，会覆盖配置文件的相关设置，此时需以代码配置为准。

内容的提问来源于stack exchange，提问作者The light one