KSQLDB推送查询反序列化失败：使用错误Schema ID查找模式

核心问题分析 #

你的Debezium连接器配置存在参数错误，导致消息未按Confluent兼容格式（带Schema ID前缀）写入Kafka主题，KSQLDB读取时无法识别注册表中的有效Schema ID，默认使用ID 0查找，从而触发404错误。

分步解决方案 #

1. 修正Debezium连接器配置 #

连接器配置中有两处关键错误：

• 参数value.apicurio.registry.as-confluent缺少converter前缀，Apicurio转换器无法识别该参数，导致消息未添加Schema ID前缀
• 冗余配置了Confluent Schema Registry的参数value.converter.schema.registry.url，与Apicurio配置冲突

修正后的连接器配置：

{
  "name": "new.connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "172.17.203.10",
    "database.port": "6000",
    "database.user": "postgres",
    "database.password": "123",
    "database.dbname": "stss_market_info",
    "database.server.name": "stss.market.info",
    "table.include.list": "public.ice_symbols",
    "message.key.columns": "public.ice_symbols:name",
    "snapshot.mode": "always",
    "transforms": "unwrap,extractKey",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "name",
    "value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
    "value.converter.apicurio.registry.url": "http://local-server:8080/apis/registry/v2",
    "value.converter.apicurio.registry.auto-register": true,
    "value.converter.apicurio.registry.find-latest": true,
    "value.converter.apicurio.registry.as-confluent": true,
    "name": "new.connector"
  }
}

修改后重启连接器，重新生成主题消息。

2. 配置KSQLDB指向Apicurio Registry #

在KSQLDB的ksql-server.properties文件中添加以下配置，确保KSQLDB使用Apicurio进行Avro反序列化：

# 指定Apicurio Registry地址
ksql.schema.registry.url=http://local-server:8080/apis/registry/v2
ksql.confluent.schema.registry.url=http://local-server:8080/apis/registry/v2
ksql.avro.schema.registry.url=http://local-server:8080/apis/registry/v2

# 默认序列化/反序列化格式为AVRO
ksql.serialization.format.default=AVRO

重启KSQLDB服务器使配置生效。

3. 重新创建KSQLDB表并验证 #

删除旧表后重新创建，明确指定VALUE_FORMAT为AVRO：

DROP TABLE IF EXISTS ICE_SYMBOLS;
CREATE TABLE ICE_SYMBOLS (
    name VARCHAR PRIMARY KEY,
    -- 根据实际表结构补充其他字段
) WITH (
    KAFKA_TOPIC='stss.market.info.public.ice_symbols',
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='KAFKA'
);

执行查询验证数据：

SELECT * FROM ICE_SYMBOLS EMIT CHANGES LIMIT 10;

4. 验证消息格式（可选） #

用Kafka命令行工具检查消息是否包含Schema ID前缀：

kafka-console-consumer --bootstrap-server <你的Kafka Broker地址>:9092 --topic stss.market.info.public.ice_symbols --from-beginning --property print.key=true

若消息开头存在4字节非打印字符，说明格式正确，KSQLDB可正常解析Schema ID。

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