怎么避免flink sql cdc作业重启后重新从头开始消费binlog?

[casel.chen]

我有一个如下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'
);