使用@KafkaListener类级别注解与@KafkaHandler处理多类型JSON消息时的反序列化及类型映射问题

看起来你在跨项目处理Kafka多类型JSON消息消费时，碰到了典型的反序列化和类型映射问题，我来帮你梳理下问题根源和解决步骤：

核心问题分析 #

你遇到的两个错误本质上是连锁反应：

1. 最初的IllegalArgumentException是因为Kafka的JSON反序列化器默认只信任JDK自带的包，你通过设置TRUSTED_PACKAGES="*"解决了这个问题，但随即出现了ClassNotFoundException。
2. 这个类找不到的问题，是因为生产者一开始把自己项目里的类全限定名（比如br.com.producer.chave.model.Chave）写入了消息的类型头中，而消费者项目里并没有这个包路径的类，反序列化器自然找不到。
3. 你尝试配置TYPE_MAPPINGS但没生效，大概率是因为生产者端的类型别名映射没有正确作用到消息序列化过程，导致消息头里还是带的全类名，消费者的映射规则匹配不上。

分步解决方案 #

1. 完善生产者端的JSON序列化配置 #

确保生产者在序列化多类型消息时，用别名替代全类名写入消息头，这样消费者就能通过别名映射到自己项目的类：

@Bean
public ProducerFactory<String,Object> producerFactory() {
    JsonSerializer serializer = new JsonSerializer();
    Map<String, Object> configProps = new HashMap<>();
    // 配置类型别名映射：生产者类 -> 别名
    configProps.put(JsonSerializer.TYPE_MAPPINGS,
        "chave:br.com.producer.chave.model.Chave," +
        "cpfCnpj:br.com.producer.cpfcnpj.model.CpfCnpj");
    // 显式开启类型头（默认是true，但显式配置更稳妥）
    configProps.put(JsonSerializer.USE_TYPE_INFO_HEADERS, true);
    serializer.configure(configProps, false);
    return new DefaultKafkaProducerFactory<>(configProperties(), new StringSerializer(), serializer);
}

2. 修正消费者端的JSON反序列化配置 #

消费者需要对应生产者的别名，映射到自己项目的实体类，同时确保读取消息头里的别名信息：

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    JsonDeserializer deserializer = new JsonDeserializer();
    Map<String, Object> configProps = new HashMap<>();
    // 配置别名 -> 消费者本地类的映射，和生产者的别名完全对应
    configProps.put(JsonDeserializer.TYPE_MAPPINGS,
        "chave:br.com.consumer.psib.domain.model.Chave," +
        "cpfCnpj:br.com.consumer.psib.domain.model.CpfCnpj");
    configProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
    // 显式开启读取类型头
    configProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, true);
    deserializer.configure(configProps, false);
    // 移除第四个参数，用默认的三参数构造器即可
    return new DefaultKafkaConsumerFactory<>(configProperties(), new StringDeserializer(), deserializer);
}

重要修正：你的消费者configProperties()方法遗漏了安全相关的配置参数（SSL、SASL这些），需要补充进去，否则可能连不上Kafka集群：

private Map<String, Object> configProperties() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
    configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.valueOf(autoCommit));
    // 补充安全配置
    configProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, sslSecurityProtocol);
    configProps.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
    configProps.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
    configProps.put(SslConfigs.SSL_PROTOCOL_CONFIG, sslProtocol);
    configProps.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
    configProps.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, sslTruststorePassword);
    configProps.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, sslTruststoreType);
    configProps.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
    return configProps;
}

3. 验证监听类与实体类 #

• 确保消费者项目里的Chave和CpfCnpj类的字段名、类型和生产者的对应类完全一致（JSON反序列化依赖字段匹配）。
• 你的@KafkaHandler方法配置是正确的，默认方法也能捕获无法匹配的消息，记得在方法里调用acknowledgment.acknowledge()手动提交偏移量（因为你设置了MANUAL Ack模式）。

4. 重新发送测试消息 #

之前发送的旧消息可能还是带有生产者的全类名类型头，这些消息在消费者端依然会报错，所以需要重新发送一批用新配置生成的测试消息来验证。

额外注意事项 #

• 如果需要兼容旧消息（带全类名的），可以在消费者的TYPE_MAPPINGS里额外添加生产者全类名到消费者类的映射，比如："br.com.producer.chave.model.Chave:br.com.consumer.psib.domain.model.Chave"
• 尽量避免使用TRUSTED_PACKAGES="*"，在生产环境可以指定消费者自己的实体类包路径，比如"br.com.consumer.psib.domain.model"

备注：内容来源于stack exchange，提问作者Henrique Mendes