Spring Kafka消费者反序列化自定义事件类时触发信任包异常

核心问题分析 #

1. 信任包配置错误：你将全类名填入TRUSTED_PACKAGES，但该配置要求的是包名而非具体类名——即使日志显示类在信任列表中，实际校验逻辑是按包路径匹配的。
2. 缺少ErrorHandlingDeserializer：日志明确提示需要配置该类处理序列化异常，默认错误处理器无法直接处理SerializationException。
3. 泛型类型不明确：消费者工厂使用Object作为泛型，导致JsonDeserializer无法确定目标反序列化类型，增加了类型映射风险。

分步修复 #

步骤1：修正信任包配置

将消费者配置中的TRUSTED_PACKAGES值改为包名，而非全类名：

public Map<String, Object> consumerConfig() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    
    // 替换原错误的全类名配置为包名
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.microservice.provider.persistence.dto");
    // 测试环境可临时用*信任所有包（生产环境不建议）
    // props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
    
    return props;
}

步骤2：配置ErrorHandlingDeserializer

用ErrorHandlingDeserializer包裹JsonDeserializer，让框架能正确处理序列化异常：

public Map<String, Object> consumerConfig() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    
    // 配置顶层ErrorHandlingDeserializer
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    // 指定底层实际使用的JsonDeserializer
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
    
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.microservice.provider.persistence.dto");
    return props;
}

步骤3：明确消费者工厂泛型类型

将消费者工厂的泛型从Object改为具体的OrderPlacedEvent，直接配置JsonDeserializer实例避免配置传递歧义：

@Bean
public ConsumerFactory<String, OrderPlacedEvent> consumerFactory(){
    JsonDeserializer<OrderPlacedEvent> jsonDeserializer = new JsonDeserializer<>(OrderPlacedEvent.class);
    jsonDeserializer.addTrustedPackages("com.microservice.provider.persistence.dto");
    return new DefaultKafkaConsumerFactory<>(
            consumerConfig(),
            new StringDeserializer(),
            jsonDeserializer
    );
}

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, OrderPlacedEvent>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, OrderPlacedEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

步骤4：修正消费者监听方法参数

避免使用Object作为参数名（Java关键字，不规范），改为有意义的名称：

@KafkaListener(
        topics = {"providerconsumer"},
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "my-group-id"
)
public void listener(OrderPlacedEvent event) {
    log.info("Mensaje recibido el mensaje es : " + event.getNombre());
}

额外检查（可选） #

如果生产者和消费者的OrderPlacedEvent类路径不一致，需在双方配置中添加类型映射：

// 生产者配置添加
properties.put(JsonSerializer.TYPE_MAPPINGS, "orderPlacedEvent:com.microservice.provider.persistence.dto.OrderPlacedEvent");

// 消费者配置对应添加
props.put(JsonDeserializer.TYPE_MAPPINGS, "orderPlacedEvent:com.microservice.provider.persistence.dto.OrderPlacedEvent");

内容的提问来源于stack exchange，提问作者Steeven Rodriguez Zhunio