StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用 checkpoint,设置触发间隔（两次执行开始时间间隔）
env.enableCheckpointing(3000);
// 模式支持EXACTLY_ONCE()/AT_LEAST_ONCE()
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 存储位置，FileSystemCheckpointStorage(文件存储)
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///c:/cdc/checkpoint/"));
// savepoint存储位置
// env.setDefaultSavepointDirectory("file:///c:/cdc/checkpoint/");
// 超时时间，checkpoint没在时间内完成则丢弃
env.getCheckpointConfig().setCheckpointTimeout(10000L);
// 同时并发数量
env.getCheckpointConfig().setMaxConcurrentCheckpoints(2); 
// 最小间隔时间（前一次结束时间，与下一次开始时间间隔）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000); 
// 外部checkpoint(例如文件存储)清除策略
env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);