Quarkus Streams中Debezium消息序列化反序列化问题排查

1. 修正POJO的注解映射 #

Debezium CDC消息里的op字段嵌套在payload节点下，你的POJO必须严格匹配字段名，必须添加@JsonProperty注解明确映射关系，避免Jackson自动命名策略出错：

import com.fasterxml.jackson.annotation.JsonProperty;

// 顶层Debezium消息类，对应完整的CDC消息结构
public class DebeziumMessage {
    private Payload payload;

    public Payload getPayload() { return payload; }
    public void setPayload(Payload payload) { this.payload = payload; }
}

public class Payload {
    @JsonProperty("op") // 强制绑定Debezium消息中的小写op字段
    private String op;
    private After after;
    private Before before;

    public String getOp() { return op; }
    public void setOp(String op) { this.op = op; }
    // 其他字段的getter/setter
}

public class After {
    // 业务字段需根据Debezium消息实际结构添加@JsonProperty
}

public class Before {
    // 业务字段需根据Debezium消息实际结构添加@JsonProperty
}

2. 优化Quarkus应用配置 #

在application.properties中配置适配Debezium的序列化规则，推荐用Quarkus内置的JsonSerde减少自定义成本：

# Kafka Streams基础配置
quarkus.kafka-streams.default.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
quarkus.kafka-streams.default.value-serde=io.quarkus.kafka.streams.runtime.serialization.JsonSerde

# Jackson兼容配置
quarkus.jackson.deserialization.fail-on-unknown-properties=false # 忽略消息中POJO未定义的冗余字段
quarkus.jackson.mapper-feature.ACCEPT_CASE_INSENSITIVE_PROPERTIES=true # 兼容字段大小写差异（可选）

3. 简化Kafka Streams拓扑构建 #

无需自定义Serializer/Deserializer，直接用JsonSerde绑定顶层消息类即可：

import io.quarkus.kafka.streams.runtime.serialization.JsonSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Produces;

@ApplicationScoped
public class KafkaStreamTopology {

    @Produces
    public KStream<String, DebeziumMessage> buildTopology(StreamsBuilder builder) {
        JsonSerde<DebeziumMessage> messageSerde = new JsonSerde<>(DebeziumMessage.class);
        
        KStream<String, DebeziumMessage> stream = builder.stream(
            "your-debezium-topic",
            Consumed.with(Serdes.String(), messageSerde)
        );

        stream.foreach((key, msg) -> {
            if (msg.getPayload() != null && msg.getPayload().getOp() != null) {
                // 正常执行业务逻辑
                System.out.println("操作类型: " + msg.getPayload().getOp());
            } else {
                System.err.println("Payload或op字段为空");
            }
        });

        return stream;
    }
}

4. 自定义Serde的修复方案（若坚持使用） #

如果一定要用自定义的PayloadSerializer/Deserializer，必须先解析完整的Debezium消息结构，再提取payload：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

public class PayloadDeserializer implements Deserializer<Payload> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public Payload deserialize(String topic, byte[] data) {
        if (data == null) return null;
        try {
            // 先解析顶层消息，再取出payload，不要直接解析Payload类
            DebeziumMessage fullMessage = objectMapper.readValue(data, DebeziumMessage.class);
            return fullMessage.getPayload();
        } catch (Exception e) {
            throw new RuntimeException("反序列化Payload失败", e);
        }
    }

    @Override
    public void close() {}
}

关键排查点 #

• 用kafka-console-consumer查看Debezium消息的实际结构，确认op字段确实在payload节点下，字段名为小写op
• 开启Jackson debug日志，查看反序列化时是否有字段匹配失败的提示
• 检查POJO字段名是否和消息字段名完全匹配（比如Java字段用operation而不是op，却未添加@JsonProperty("op")）

内容的提问来源于stack exchange，提问作者André Martins