Kafka发送嵌套JSON对象异常：消费者仅接收根级字段问题

问题描述 #

搭建基础Kafka项目时，发送包含嵌套List<Order>结构的UserCreate对象，消费者接收后仅根级createDate字段有值，嵌套的orders字段始终为null。

相关代码 #

UserCreate类

@Builder
@Jacksonized
@ToString
public class UserCreate implements Serializable {

    @JsonSerialize(using = LocalDateSerializer.class)
    @JsonDeserialize(using = LocalDateDeserializer.class)
    private final LocalDate createDate;

    private final List<Order> orders;

}

Order类

@Builder
@Jacksonized
@ToString
public class Order implements Serializable {
   private final String name;
}

生产者配置

@Bean
public ProducerFactory<String, UserCreate> transferProducerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, TransferRequest> transferKafkaTemplate() {
    return new KafkaTemplate<>(transferProducerFactory());
}

消费者配置

public ConsumerFactory<String, UserCreate> transferConsumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group-id");
    return new DefaultKafkaConsumerFactory<>(
        props,
        new StringDeserializer(),
        new JsonDeserializer<>(UserCreate.class));
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, UserCreate> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, UserCreate> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(transferConsumerFactory());
    return factory;
}

接收结果

Received message: UserCreate(createDate=2023-10-08, orders=null)

修复方案 #

1. 修正生产者KafkaTemplate泛型错误 #

生产者配置中KafkaTemplate的泛型错误地写为TransferRequest，但实际发送的是UserCreate对象，这会导致序列化逻辑混乱，嵌套字段无法正确被序列化。

修改后的代码：

@Bean
public KafkaTemplate<String, UserCreate> transferKafkaTemplate() {
    return new KafkaTemplate<>(transferProducerFactory());
}

2. 配置类型映射与信任包 #

JsonDeserializer默认无法自动识别嵌套类的类型信息，需要在生产者和消费者两端配置类型映射，同时开启信任包避免反序列化权限问题。

修改后的生产者配置

@Bean
public ProducerFactory<String, UserCreate> transferProducerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    // 添加类型映射，替换为你的实体类全限定名
    configProps.put(JsonSerializer.TYPE_MAPPINGS, 
        "UserCreate:com.yourpackage.UserCreate,Order:com.yourpackage.Order");
    return new DefaultKafkaProducerFactory<>(configProps);
}

修改后的消费者配置

public ConsumerFactory<String, UserCreate> transferConsumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group-id");
    // 与生产者一致的类型映射
    props.put(JsonDeserializer.TYPE_MAPPINGS, 
        "UserCreate:com.yourpackage.UserCreate,Order:com.yourpackage.Order");
    // 信任实体类所在的包
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.yourpackage");
    return new DefaultKafkaConsumerFactory<>(
        props,
        new StringDeserializer(),
        new JsonDeserializer<>(UserCreate.class));
}

3. 确认实体类Jackson配置有效性 #

确保@Jacksonized与@Builder注解正确配合，Jackson能够处理不可变类的反序列化。如果仍有问题，可以尝试为Order类添加@JsonIgnoreProperties(ignoreUnknown = true)，避免因未知字段导致反序列化失败。

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