如何让Flink在批处理模式下将.inprogress文件重命名为part-xxx？

你遇到的核心问题是批处理模式下检查点自动禁用导致的测试验证困扰，其实不用非得采用无界源加单独线程强制终止的繁琐方案，这里有几个更简洁的处理方式：

1. 利用批处理作业的自动结束特性 #

因为有界源触发的是Flink批处理模式，作业会在所有数据处理完成后自动终止。单元测试里直接提交作业并等待完成即可，之后再去校验输出的Parquet文件：

// 构建作业逻辑
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// ... 数据源定义、数据转换、Parquet写入逻辑 ...

// 提交作业并等待完成
JobExecutionResult result = env.execute("Bounded Source to Parquet Job");
result.waitForCompletion();

// 此处开始校验输出Parquet文件的内容、数量等

这种方式完全不需要额外线程，作业跑完自然结束，是最直接的解决方案。

2. 强制切换为流处理模式（按需使用） #

如果你的测试场景必须依赖检查点机制，可以显式将执行模式设置为STREAMING，这样检查点会被启用，ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH也能生效：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 强制设置为流处理模式
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// 启用检查点
env.enableCheckpointing(1000);
// 配置任务完成后触发检查点
Configuration config = new Configuration();
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
env.configure(config);

// ... 后续作业逻辑 ...

这种模式下，有界源处理完数据后作业会进入空闲状态，你可以通过设置env.setRestartStrategy(RestartStrategies.noRestart())避免自动重启，或者在测试中等待检查点完成后手动取消作业。

3. 自定义文件状态检查逻辑 #

如果担心批处理作业出现异常卡住，可以封装一个简单的工具方法：在提交作业后，定期检查输出目录的Parquet文件状态（比如是否生成了Flink写Parquet时的.finished标记文件，或者文件大小在一段时间内不再变化），满足条件就结束等待，无需依赖无界源。

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