在上一篇文章《Flink kafka source源码解析:非checkpoint模式下offset的提交》中讲了非checkpoint模式下，offset要么使用kafka的自动提交模式，要么完全禁止提交offset。
本文接着将checkpoint模式下，offset提交。
初始化offsetCommitMode
通过上文可以知道，当调用了env.enableCheckpointing方法后 offsetCommitMode 的值就是ON_CHECKPOINTS，而且会通过下面方法强制关闭 kafka 自动提交功能，这个值很重要，后续很多地方都是根据这个值去判断如何操作的。这个方法在org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase#run中调用createFetcher，createFetcher中调用以下方法adjustAutoCommitConfig：
protected static void adjustAutoCommitConfig(Properties properties, OffsetCommitMode offsetCommitMode