Kafka KSQLDB Stream无法读取user_event主题问题求助

我有一个向user_event主题生产消息的系统，已通过以下语句创建对应的KSQLDB流：

CREATE STREAM UserEventStream (
    type STRING,
    bundle_version BIGINT,
    app_version STRING,
    os_name STRING,
    device_id STRING,
    session_id STRING,
    device_type STRING,
    device_name STRING,
    device_brand STRING,
    language STRING,
    ip STRING,
    user_agent STRING,
    user_id BIGINT,
    fcm_token STRING,
    iframe BOOLEAN,
    referrer STRING,
    data MAP<STRING, STRING>,
    time BIGINT
) WITH (
    KAFKA_TOPIC='user_event',
    VALUE_FORMAT='AVRO'
);

执行SELECT * FROM UserEventStream;查询时无结果返回，但该主题每秒都有新数据产生。消息格式确认正确，一周前该流还能正常读取数据，期间未修改任何系统或服务器。

Kafka、Schema Registry、KSQLDB、kafka-ui均部署在非Docker虚拟机上，对应的AVRO消息Schema如下：

{
    "type": "record",
    "name": "UserEvent",
    "namespace": "com.mycompany.kafka.records",
    "fields": [
        {
            "name": "type",
            "type": {
                "type": "string",
                "avro.java.string": "String"
            }
        },
        {
            "name": "bundle_version",
            "type": [
                "null",
                "long"
            ]
        },
        {
            "name": "app_version",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "os_name",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "device_id",
            "type": {
                "type": "string",
                "avro.java.string": "String"
            }
        },
        {
            "name": "session_id",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "device_type",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "device_name",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "device_brand",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "language",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "ip",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "user_agent",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "user_id",
            "type": [
                "null",
                "long"
            ]
        },
        {
            "name": "fcm_token",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "iframe",
            "type": "boolean"
        },
        {
            "name": "referrer",
            "type": [
                "null",
                {
                    "type": "string",
                    "avro.java.string": "String"
                }
            ]
        },
        {
            "name": "data",
            "type": {
                "type": "map",
                "values": [
                    "null",
                    {
                        "type": "string",
                        "avro.java.string": "String"
                    }
                ],
                "avro.java.string": "String"
            },
            "default": {}
        },
        {
            "name": "time",
            "type": {
                "type": "long",
                "logicalType": "timestamp-millis"
            }
        }
    ]
}

1. 调整流的起始消费位置 #

KSQLDB默认从主题最新偏移量开始消费，若查询时新消息还未产生，或流创建早于现有消息，会导致无结果返回。可以：

• 重新创建流时指定从最早位置消费：

CREATE STREAM UserEventStream (
    -- 字段定义保持不变
) WITH (
    KAFKA_TOPIC='user_event',
    VALUE_FORMAT='AVRO',
    START_OFFSET='earliest'
);

• 或修改现有流的消费起始位置：

ALTER STREAM UserEventStream SET (START_OFFSET='earliest');

2. 修正流字段与AVRO Schema的兼容性 #

AVRO Schema中部分字段为可空类型，但KSQLDB流定义为非空，若消息中存在null值会被丢弃：

• 将bundle_version和user_id改为可空类型：

bundle_version BIGINT NULL,
user_id BIGINT NULL,

• data字段的AVRO值允许null，需同步调整KSQLDB定义：

data MAP<STRING, STRING NULL>,

3. 检查服务连接与权限 #

• 查看KSQLDB日志，确认是否能正常连接Kafka集群和Schema Registry，有无权限报错。
• 验证KSQLDB使用的账号是否拥有user_event主题的读取权限，以及Schema Registry的Schema读取权限。

4. 直接验证消息内容 #

• 用kafka-ui查看user_event主题的消息，确认Schema ID与Registry中注册的一致，内容符合定义。
• 使用命令行工具直接消费验证：

kafka-avro-console-consumer --bootstrap-server <kafka-broker>:9092 --topic user_event --from-beginning --property schema.registry.url=http://<schema-registry>:8081

若该命令能正常输出消息，问题出在KSQLDB配置或流定义；若也无法解析，说明Schema Registry或消息本身存在异常。

5. 重启KSQLDB服务 #

排查KSQLDB进程状态，若存在临时连接或缓存问题，尝试重启服务后重新查询。

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