我有一个如下flink sql cdc作业,设置了'scan.startup.mode' = 'latest-offset'。但在作业重启后发现它又从头开始消费binlog,导致sink下游数据库频繁报duplicate key error,有什么办法可以避免吗?
CREATE TABLE `mysql_source` ( `id` STRING, `acct_id` STRING, `acct_name` STRING, `acct_type` STRING, `acct_bal` STRING, PRIMARY KEY (`id`) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'mysql', 'password' = 'mysql', 'database-name' = 'test', 'scan.startup.mode' = 'latest-offset', 'table-name' = 'test', 'server-time-zone' = 'Asia/Shanghai' );