Spring Boot DDD微服务Kafka消费事件抛出content为null异常

核心报错原因 #

你遇到的argument "content" is null异常，本质是Kafka消费者拿到的消息payload为空，调用DomainEvent.fromJson时传入了null值触发Jackson的参数校验失败。

具体问题点与修复步骤 #

1. 修复DomainEvent类toJson方法吞异常的问题 #

当前toJson方法捕获序列化异常后没有任何处理，序列化失败直接返回null，导致Kafka发送空消息。同时Java 8的Instant类型默认未注册序列化模块时会触发序列化失败，刚好被空catch块吞掉。
修改后的代码：

public String toJson() {
    ObjectMapper objectMapper = new ObjectMapper();
    // 注册Java 8时间类型序列化支持，解决Instant序列化失败问题
    objectMapper.registerModule(new JavaTimeModule());
    try {
        return objectMapper.writeValueAsString(this);
    } catch (JsonProcessingException e) {
        // 抛出异常避免静默失败，早发现问题
        throw new RuntimeException("领域事件序列化失败", e);
    }
}

注意：Spring Boot项目默认spring-boot-starter-web已经包含JSR310序列化模块，非Spring Boot项目可手动引入com.fasterxml.jackson.datatype:jackson-datatype-jsr310依赖

2. 修复订单审批逻辑硬编码orderId的问题 #

当前发布事件时写死了固定orderId，没有使用当前审批通过的订单ID，后续即使消费成功也无法匹配到对应宠物：

// 原有错误代码
domainEventPublisher.publish(new OrderApproved("e49b4057-582b-42fa-beed-e3b9e6811cdc"));
// 修改为动态获取当前订单ID
domainEventPublisher.publish(new OrderApproved(order.getId().getId()));

注：根据你的代码结构，Order的主键是封装类OrderId，调用getId()即可拿到实际的字符串ID

3. 消费者增加payload非空校验 #

避免空payload进入反序列化逻辑导致无意义报错：

@KafkaListener(topics = TopicHolder.TOPIC_ORDER_APPROVED, groupId = "petCatalog")
public void consumeOrderApproved(@Payload(required = false) String jsonMessage){
    try{
        System.out.println("Topic consumed successfully");
        // 新增非空校验
        if (jsonMessage == null || jsonMessage.isBlank()) {
            System.out.println("收到空消息，跳过处理");
            return;
        }
        OrderApproved event = DomainEvent.fromJson(jsonMessage, OrderApproved.class);
        System.out.println("Print after creating an event");
        petService.updatePet(event.getOrderId());
        System.out.println("event.getOrderId(): "+ event.getOrderId());
    } catch (Exception e){
        e.printStackTrace();
    }
}

4. 可选优化：复用ObjectMapper实例 #

每次调用toJson/fromJson都新建ObjectMapper是不必要的，建议改为全局单例复用，降低性能开销：

@Getter
public class DomainEvent {
    // 全局复用配置好的ObjectMapper
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
            .registerModule(new JavaTimeModule());
    
    private String topic;
    private Instant occurredOn;
    public DomainEvent(String topic) {
        this.occurredOn = Instant.now();
        this.topic = topic;
    }
    public String toJson() {
        try {
            return OBJECT_MAPPER.writeValueAsString(this);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("领域事件序列化失败", e);
        }
    }
    public static <E extends DomainEvent> E fromJson(String json, Class<E> eventClass) throws JsonProcessingException {
        return OBJECT_MAPPER.readValue(json,eventClass);
    }
    // 其余原有方法保持不变
}

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