Confluent ksqlDB 无法基于users主题创建stream问题求助

前置排查步骤 #

• 执行ksqlDB命令SHOW TOPICS;确认users主题已存在，且ksqlDB服务持有该主题的CONSUME、DESCRIBE权限
• 执行命令PRINT 'users' FROM BEGINNING LIMIT 10;获取users主题的真实消息格式，确认消息序列化类型与创建语句中指定的格式一致，可选格式包含AVRO、JSON、PROTOBUF等
• 若使用AVRO/PROTOBUF等需要Schema Registry的格式，确认ksqlDB配置项ksql.schema.registry.url指向正确的Schema Registry服务地址，且服务连通性正常

标准创建语句示例 #

AVRO格式显式指定Subject示例 #

CREATE STREAM users_stream (
    user_id INT,
    username STRING,
    email STRING,
    register_ts BIGINT
) WITH (
    KAFKA_TOPIC='users',
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='KAFKA',
    -- 显式指定Schema Registry对应的subject名称
    VALUE_SUBJECT_NAME='users-value',
    KEY_SUBJECT_NAME='users-key'
    -- 无对应schema时可开启自动注册配置
    -- KSQL_SCHEMA_REGISTRY_AUTO_REGISTER_SCHEMAS=true
);

JSON格式无Schema场景示例 #

CREATE STREAM users_stream (
    user_id INT,
    username STRING,
    email STRING,
    register_ts BIGINT
) WITH (
    KAFKA_TOPIC='users',
    VALUE_FORMAT='JSON',
    KEY_FORMAT='KAFKA'
);

常见报错修复方案 #

报错提示Subject not found：检查Schema Registry中是否存在你指定的users-value/users-key对应subject，不存在可以手动注册schema，也可以在WITH参数中添加KSQL_SCHEMA_REGISTRY_AUTO_REGISTER_SCHEMAS=true开启自动注册
报错提示Topic not found：提前在kafka集群创建users主题，或者在CREATE语句后添加IF NOT EXISTS参数
报错提示Serialization mismatch：核对PRINT命令输出的消息字段类型和创建语句中定义的字段类型是否一致，调整字段定义匹配真实消息结构即可

参考操作截图 #

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