消费Debezium产生的Kafka Avro数据时类型转换与序列化异常如何解决？

前置配置修正 #

• 弃用自定义Avro反序列化器，改用Confluent官方的KafkaAvroDeserializer，它会自动从Schema Registry拉取对应schema，无需手动处理schema匹配逻辑。配置示例如下：

spring:
  kafka:
    consumer:
      bootstrap-servers: {Kafka集群地址}
      key-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        schema.registry.url: {Schema Registry服务地址}
        specific.avro.reader: false # 先设为false用GenericRecord灵活接收，用生成类时再改为true

• @KafkaListener入参不要强制指定Key类型为Integer，Debezium默认生成的消息Key是对应表主键的Avro复合类型，和你假设的Integer不匹配，直接用ConsumerRecord<Object, GenericRecord>接收即可，不需要用到Key的话可以直接忽略Key部分。

从Schema Registry获取Debezium生成的Schema #

直接调用Schema Registry的HTTP接口拉取对应topic的最新schema，保存为avsc格式文件：

• 拉取value schema：curl {Schema Registry地址}/subjects/{Debezium同步topic名}-value/versions/latest | jq .schema > shop_value.avsc
• 拉取key schema：curl {Schema Registry地址}/subjects/{Debezium同步topic名}-key/versions/latest | jq .schema > shop_key.avsc

两种转业务POJO的方案 #

方案1：生成Avro SpecificRecord类（推荐，性能高、类型安全） #

1. 将上一步获取的avsc文件放到项目src/main/resources/avro目录下
2. 引入avro代码生成maven插件，自动生成对应的Java类：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.11.3</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/resources/avro</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

3. 将消费者配置中的specific.avro.reader改为true，调整监听方法入参为生成的类：

@KafkaListener(topics = "{你的topic名}", groupId = "{你的消费组}")
public void consume(ConsumerRecord<ShopKey, ShopValue> record) {
    ShopValue debeziumMsg = record.value();
    // 删除操作after为null，需提前判断
    if (debeziumMsg.getAfter() == null) {
        return;
    }
    // 直接通过生成类的get方法获取字段，转成业务POJO即可
    Shop shop = new Shop();
    shop.setId(debeziumMsg.getAfter().getId());
    shop.setName(debeziumMsg.getAfter().getName());
    // 其他字段赋值
}

方案2：手动映射GenericRecord（无需生成类，灵活性高） #

如果不想维护自动生成的Avro类，可以直接从GenericRecord中逐个提取字段赋值给业务POJO，不要用ObjectMapper直接转换GenericRecord，二者序列化逻辑不兼容会报错：

@KafkaListener(topics = "{你的topic名}", groupId = "{你的消费组}")
public void consume(ConsumerRecord<Object, GenericRecord> record) {
    GenericRecord value = record.value();
    GenericRecord after = (GenericRecord) value.get("after");
    if (after == null) {
        return;
    }
    Shop shop = new Shop();
    shop.setId((Long) after.get("id"));
    shop.setName((String) after.get("name"));
    // 其他字段依次赋值，字段名需和Debezium同步的表字段完全一致
}

历史报错原因说明 #

1. ShopMessage无法强制转换为org.springframework.messaging.Message：是因为自定义反序列化器返回的是Avro对象，不符合Spring Messaging的Message包装规范，直接用ConsumerRecord接收即可规避该问题。
2. 找不到bookingdev_sqip_local.booking.shop.Key类：是因为你强制指定Key类型为Integer，和Debezium生成的复合类型Key不匹配，要么用生成的Key类，要么直接用Object接收Key即可。

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