flink新版的kafkaSource API如何在执行Checkpoint的时候提交offset到Checkpoint

在Flink新版的KafkaSource API中，可以通过设置`enableCheckpointing`属性来开启Flink的Checkpoint机制。在Checkpoint时，KafkaSource会自动提交偏移量到Kafka的__consumer_offsets主题以便后续恢复。具体来说，可以按照以下步骤进行操作：

1. 在创建KafkaSource时，通过`enableCheckpointing`方法设置启用Checkpoint机制，并设置Checkpoint的间隔时间和模式，例如：

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
    "my-topic",
    new SimpleStringSchema(),
    properties);
consumer.setCommitOffsetsOnCheckpoints(true);
consumer.setStartFromEarliest();
consumer.setEnableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

2. 在执行Checkpoint时，KafkaSource会自动提交当前的消费偏移量到Kafka的__consumer_offsets主题中，以便后续的恢复。可以通过`setCommitOffsetsOnCheckpoints`方法设置是否在Checkpoint时提交偏移量，默认为true。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(10000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.addSource(consumer)
   .map(new MyMapper())
   .addSink(new MySink());
env.execute("Flink KafkaSource Example");

在上述代码中，我们开启了Flink的Checkpoint机制，并设置了Checkpoint的时间间隔为10秒，模式为EXACTLY_ONCE。同时，我们也设置了KafkaSource在Checkpoint时提交偏移量到Kafka的__consumer_offsets主题中。