Flink CDC知道这个错误是什么原因吗？

Flink CDC知道这个错误是什么原因吗？我用flink 1.17.1 sql，flink mysql cdc 2.4.2读mysql数据到starrocks？

CREATE DATABASE IF NOT EXISTS default_catalog.ride;

CREATE TABLE IF NOT EXISTS default_catalog.ride.ride_src (
id BIGINT NOT NULL,
stat BIGINT NULL,
...
PRIMARY KEY(id)
NOT ENFORCED
) with (
'port' = '3306',
'username' = 'xxxxx',
'password' = 'xxxxx',
'database-name' = 'ride',
'table-name' = 'ride',
'connector' = 'mysql-cdc',
'hostname' = 'xxxxxxx',
'server-time-zone' = 'UTC'
);

CREATE TABLE IF NOT EXISTS default_catalog.ride.ride_sink (
id BIGINT NOT NULL,
stat BIGINT NULL,
...
PRIMARY KEY(id)
NOT ENFORCED
) with (
'sink.properties.column_separator' = '\x01',
'sink.buffer-flush.interval-ms' = '15000',
'sink.properties.strip_outer_array' = 'true',
'jdbc-url' = 'jdbc:mysql://flink002:9030',
'connector' = 'starrocks',
'database-name' = 'ride',
'table-name' = 'ride',
'load-url' = 'flink002:8030',
'sink.properties.format' = 'csv',
'username' = 'xxxxx',
'sink.max-retries' = '10',
'sink.properties.row_delimiter' = '\x02',
'password' = 'xxxxxx'
);

INSERT INTO default_catalog.ride.ride_sink SELECT * FROM default_catalog.ride.ride_src;