如何为Flink批处理执行模式实现BOUNDED有界自定义数据源

问题根源 #

Flink 1.12 引入的批流一体运行模式中，RuntimeExecutionMode.BATCH 会强制校验所有数据源必须是有界（BOUNDED）类型。你使用的旧版 SourceFunction API 没有声明有界性的能力，Flink 默认会将所有 SourceFunction 实现识别为无界（UNBOUNDED）源，哪怕你当前的MongoSource逻辑是读完MongoDB全量数据就主动退出，框架层也无法感知这个特性，所以触发校验异常。

解决方案 #

方案一：最小代码改动（无需修改自定义数据源）

将执行环境的运行模式从 BATCH 改为 AUTOMATIC 即可：

env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

AUTOMATIC 模式下Flink会自动感知作业执行状态，你的MongoSource读完所有数据后会主动退出，Flink会自动按照批处理逻辑执行完全作业后正常结束，效果和BATCH模式完全一致。

方案二：改造为FLIP-27标准有界源（推荐，原生适配BATCH模式）

如果需要明确声明数据源为有界类型、完全兼容BATCH模式的校验规则，建议将自定义数据源改造为Flink 1.12推出的新Source接口（FLIP-27规范），核心实现逻辑如下：

1. 实现Source接口，声明有界性为Boundedness.BOUNDED
2. 实现对应的SplitEnumerator、SourceReader组件，复用你现有MongoDB读取逻辑即可
   示例核心代码：

public class MongoBoundedSource implements Source<Document, MongoSplit, List<MongoSplit>> {
    @Override
    public Boundedness getBoundedness() {
        // 明确声明为有界源
        return Boundedness.BOUNDED;
    }
    // 其他方法实现：拆分MongoDB集合分片、创建读取器读取对应分片数据，逻辑和你现有MongoSource的open、run方法完全一致
}

改造后执行环境保持env.setRuntimeMode(RuntimeExecutionMode.BATCH)即可正常运行。

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