如何禁用Source(from objects)->Map1链路的Checkpoint以避免中止？

针对你遇到的内存Source->Map1链路导致Checkpoint失败的问题，这里有几个实用的解决方案，你可以根据业务场景选择：

1. 直接禁用目标链路的Checkpoint参与 #

这是最贴合你需求的方案——既然你完全不关心这条链路的Checkpoint和恢复，直接让它不参与全局Checkpoint即可。Flink支持对特定DataStream链路单独禁用Checkpoint，这样全局Checkpoint触发时会忽略这条链路上的任务，不会因为它未执行而终止Checkpoint流程。

具体操作非常简单，在定义这条链路时调用disableCheckpointing()方法：

// 构建内存Source -> Map1的链路
DataStream<YourType> objectStream = env.addSource(new ObjectBasedSource())
    .map(new Map1())
    .setParallelism(1)
    .disableCheckpointing(); // 关键：禁用这条流的Checkpoint参与

// 之后正常和Kafka链路汇合到CoMap
objectStream.connect(kafkaStream)
    .map(new CoMapFunction<>())
    .addSink(new YourSink());

调用该方法后，这条链路上的所有算子都不会参与全局Checkpoint，状态也不会被保存，完全符合你“不关心该链路Checkpoint”的诉求。

2. 让内存Source持续运行（避免任务终止） #

如果不想修改Checkpoint配置，可以调整内存Source的实现：在处理完所有内存对象后，让Source进入无限等待状态（比如周期性休眠），保持任务一直处于运行状态。这样Checkpoint触发时，任务是活跃的，就不会出现“not being executed”的错误。

示例修改SourceFunction：

public class ObjectBasedSource implements SourceFunction<YourType> {
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<YourType> ctx) throws Exception {
        // 发送所有内存中的对象
        for (YourType obj : yourObjectList) {
            ctx.collect(obj);
        }
        // 处理完后进入无限等待，保持任务存活
        while (isRunning) {
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}

这个方法的缺点是会占用一个任务槽，但因为你的Map1是非并行的，资源消耗很小。

3. 拆分链路为独立作业（可选） #

如果业务允许，可以把内存Source->Map1的链路拆成一个独立的Flink作业，将处理后的结果写入临时存储（比如Kafka主题），然后原Kafka链路从这个临时主题消费数据。这样两条链路的Checkpoint完全独立，互不影响。不过这个方案需要额外的中间存储，适合对资源隔离有要求的场景。

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