【 使用环境 】生产环境 or 测试环境
【 OB or 其他组件 】
【 使用版本 】oceanbase-cdc-2.2，flink1.15.3, oblogproxy-2.0.0
【问题描述】清晰明确描述问题
【复现路径】问题出现前后相关操作
【附件及日志】
CREATE TABLE ob_tbl1 (
col1 INT,
col2 VARCHAR(20),
col3 INT)
WITH (‘connector’ = ‘oceanbase-cdc’,
‘scan.startup.mode’ = ‘initial’,
‘tenant-name’ = ‘mq_t1’,
‘username’ = ‘root’,
‘password’ = ‘pswd’,
‘database-name’ = ‘test_ob_to_kakfa’,
‘table-name’ = ‘tbl1’,
‘hostname’ = ‘127.0.0.1’,
‘port’ = ‘2881’,
‘rootserver-list’ = ‘127.0.0.1:2882:2881’,
‘logproxy.client.id’ = ‘t_user_logproxy_client_id_100004_2023111411’,
‘working-mode’ = ‘memory’,
‘logproxy.host’ = ‘127.0.0.1’,
‘logproxy.port’ = ‘2983’);
CREATE TABLE ob_tbl1_and_tbl2(
col1 INT,
col2 VARCHAR(20),
col3 INT
)WITH(
‘connector’ = ‘kafka’,
‘topic’ = ‘ob_kakfa’,
‘properties.bootstrap.servers’ = ‘10.10.101.64:9092’,
‘properties.group.id’ = ‘testGroup’,
‘scan.startup.mode’ = ‘earliest-offset’,
‘value.format’ = ‘debezium-json’
);
INSERT INTO ob_tbl1_and_tbl2 SELECT * FROM ob_tbl1;
