mdzz9527 commented on issue #3378: URL: https://github.com/apache/flink-cdc/issues/3378#issuecomment-2137045429
1. 环境 flink 1.17.2 2. 模式 yarn-session flink SQL 3. 使用flink-sql-connector-cdc-oracle-3.1.0flink-connector-jdbc-3.1.2-1.17 4. 表都有主键 oracle 的权限配置严格按照 flink cdc 文档配置 https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/flink-sources/oracle-cdc/ 5. SQL 配置如下,可以实现数据全量同步,新增数据同步,删除数据同步,无法同步数据update操作,现象:update 不会删除旧数据,同时会新增一条 ID = 0 的,且只包含变更信息的错误数据;后续测试发现:不管update 多少条数据,也只是不断更新 这条ID=0 的数据 6. SQL 如下 `SET execution.checkpointing.interval = 10min; create table products ( ID int, NAME string, DESCRIPTION string, WEIGHT float, primary key (ID) not enforced ) with ( 'connector' = 'oracle-cdc', 'hostname' = '10.2.0.79', 'port' = '1521', 'username' = 'flinkuser', 'password' = 'flinkpw', 'database-name' = 'ORCL', 'schema-name' = 'FLINKUSER', 'table-name' = 'PRODUCTS', 'debezium.log.mining.strategy' = 'online_catalog', 'debezium.log.mining.continuous.mine' = 'true', 'connection.pool.size' = '20', 'connect.max-retries' = '3', 'scan.snapshot.fetch.size' = '1024', 'scan.incremental.snapshot.chunk.size' = '8096', 'scan.incremental.snapshot.enabled' = 'true', 'debezium.snapshot.mode' = 'initial' -- initial -- schema_only ); create table tg_products ( ID decimal(9, 0), NAME string, DESCRIPTION string, WEIGHT float, primary key (ID) not enforced ) with ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://1xxxxxxxxxx:33061/transit_data_oms', 'username' = 'xxxxxxxxx', 'password' = 'xxxxxx', 'table-name' = 'products' ); insert into tg_products select * from products; ` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
