经过一段时间的尝试和阅读文档，我发现了这个问题。
田野 id 因为我的主题中的消息类型是？ INT 根据ksql表文档，它们必须是 VARCHAR 所以我按照这里描述的步骤来解决这个问题，现在一切正常。
具体步骤如下：

-- Create a stream on the original topic
CREATE STREAM users_with_wrong_key_format (userid INT, username VARCHAR, email VARCHAR)
WITH (KAFKA_TOPIC='users', VALUE_FORMAT='JSON');

-- Derive a new stream with the required key changes.
-- 1) The CAST statement converts the key to the required format.
-- 2) The PARTITION BY clause re-partitions the stream based on the new, converted key.
CREATE STREAM users_with_proper_key
WITH(KAFKA_TOPIC='users-with-proper-key') AS
SELECT CAST(userid as VARCHAR) as userid_string, username, email
FROM users_with_wrong_key_format
PARTITION BY userid_string;

-- Now you can create the table on the properly keyed stream.
CREATE TABLE users_table (userid_string VARCHAR, username VARCHAR, email VARCHAR)
WITH (KAFKA_TOPIC='users-with-proper-key',
        VALUE_FORMAT='JSON',
        KEY='userid_string');

-- Now you can create the table on the properly keyed stream.
CREATE TABLE users_table (userid_string VARCHAR, username VARCHAR, email VARCHAR)
WITH (KAFKA_TOPIC='users-with-proper-key',
    VALUE_FORMAT='JSON',
    KEY='userid_string');