Flink使用Collection源时不做Checkpoint，BucketingSink文件处于pending状态

我之前踩过一模一样的坑！这问题本质是Collection数据源的有界特性和Flink的运行模式导致的——当你用Collection生成测试数据时，Flink默认会把它当作批处理任务来处理，而批处理模式下Checkpoint机制是被默认禁用的；但当数据源是S3（尤其是读取持续新增的文件时），Flink会识别为无界流，自然会正常触发Checkpoint。

下面给你几个可行的解决方案：

1. 强制切换为流处理模式 #

这是最直接的解决办法，让Flink把Collection数据源当作无界流来处理，这样Checkpoint就会按时触发。你只需要在代码里添加一行设置运行模式的代码：

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setMaxParallelism(128)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// 关键：强制设置为流处理模式
env.setRuntimeMode(RuntimeExecutionMode.STREAMING)
env.enableCheckpointing(2000L)
// 额外配置：确保Checkpoint的精确一次语义，以及存储位置（比如S3）
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
env.getCheckpointConfig.setCheckpointStorage("s3://your-checkpoint-storage-path")

设置完之后，即使是Collection这种有界数据源，Flink也会按照流处理的逻辑触发Checkpoint，你的S3 Sink也会在Checkpoint完成后正常提交文件，让文件处于完成状态。

2. 配置S3 Sink的Commit策略 #

除了运行模式，还要确保你的S3 Sink是基于Checkpoint来提交文件的。如果你用的是StreamingFileSink，一定要配置正确的滚动策略和提交逻辑：

val sink = StreamingFileSink.forRowFormat(
  new Path("s3://your-output-path"),
  new SimpleStringEncoder[String]("UTF-8")
)
.withBucketAssigner(new DateTimeBucketAssigner[String]("yyyy-MM-dd--HH"))
.withRollingPolicy(
  DefaultRollingPolicy.builder()
    .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
    .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
    .withMaxPartSize(1024 * 1024 * 1024)
    .build()
)
.build()

这个配置下，只有当Checkpoint完成时，Sink才会把临时文件转为完成状态，避免出现未提交的文件。

3. 验证Checkpoint是否正常触发 #

你可以查看Flink TaskManager的日志，搜索关键词Triggering checkpoint或者Completed checkpoint，如果能看到这些日志，说明Checkpoint已经正常工作了。如果还是没有，检查一下是否开启了Checkpoint的日志级别（可以设置为DEBUG来查看更详细的信息）。

内容的提问来源于stack exchange，提问作者Josh Lemer