每次重新启动应用程序时重复执行RichSinkFunction#invoke方法，而且数据还是之前的数据

[zjf88865]

每次重新启动应用程序时重复执行RichSinkFunction#invoke方法，而且数据还是之前的数据，我希望我消费过了mysql修改的数据就不再去执行了，不然一堆重复数据，有没有一种机制就是我每次消费完成后有个commit的动作，表示该数据我已经消费过了，就像Canal 那样，我看大家都在吹说能支持分布式部署，这块如何操作？
这个是我的pom文件代码

<properties>
        <flink.version>1.18.1</flink.version>
    </properties>
    <dependencies>
        <!--Flink connector连接器基础包-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!--Flink CDC MySQL源-->
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-sql-connector-mysql-cdc</artifactId>
            <version>3.0.1</version>
        </dependency>
        <!--Flink的DataStream数据流API-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!--Flink的Java 客户端-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!--开启WebUI支持,端口8081,默认没有开启-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!--Flink的Table API&SQL程序可以连接到其他外部系统，用于读写批处理表和流式表。-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.11</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>

我的数据库是mysql 5.7采用gtid模式，binglog为row
Here is my code

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;




public class CustomSink extends RichSinkFunction<String> {

    private static final long serialVersionUID = -6830283501298742046L;

    @Override
    public void invoke(String json, Context context) throws Exception {
        //这一块代码一重启就重复消费之前的记录
        System.out.println(">>>" + json);
    }

    @Override
    public void  open(Configuration parameters) throws Exception {
        super.open(parameters);
        System.out.println(parameters);
    }

    @Override
    public void close() throws Exception {
        System.out.println("close");
    }

}

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MySqlSourceExample {
    public static void main(String[] args) throws Exception {
        MySqlSource<String> source = MySqlSource.<String>builder()
                .hostname("192.168.111.69")
                .port(3306)
                .databaseList("haha") // set captured database
                .tableList("haha.test") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .includeSchemaChanges(true)
                .build();
        // 启动一个webUI，指定本地WEB-UI端口号
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);
        //检查点间隔时间
        //checkpoint的侧重点是“容错”，即Flink作业意外失败并重启之后，能够直接从早先打下的checkpoint恢复运行，且不影响作业逻辑的准确性。
        env.enableCheckpointing(3000);
        // 模式支持EXACTLY_ONCE()/AT_LEAST_ONCE()
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 存储位置，FileSystemCheckpointStorage(文件存储)
        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///d:/data/log/cdc/checkpoint/"));
        // savepoint存储位置
        // env.setDefaultSavepointDirectory("file:///d:/cdc/checkpoint/");
        // 超时时间，checkpoint没在时间内完成则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        // 同时并发数量
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 最小间隔时间（前一次结束时间，与下一次开始时间间隔）
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        // 外部checkpoint(例如文件存储)清除策略
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        // 如果job失败后恢复，从最近的检查点恢复job。
//        env.getCheckpointConfig().enableApproximateLocalRecovery(true);
        env.setStateBackend(new FsStateBackend("file:///d:/data/log/cdc/stateBackend/"));
//                env.setDefaultSavepointDirectory("file:///d:/cdc/data/log/cdc");
        DataStreamSink<String> sink = env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .addSink(new CustomSink());
        env.execute();
    }
}