Flink多数据源集成测试：如何控制事件发射顺序

要在全作业集成测试中实现“第一个数据源发完所有数据后，第二个再开始发射”的需求，核心思路是通过同步机制协调两个自定义SourceFunction的执行时机，下面是具体的可落地方案：

1. 实现带同步控制的通用SourceFunction #

我们可以封装一个通用的ControlledSequentialSourceFunction，它接收待发射的数据列表，以及一个用于等待前置信号的CountDownLatch（如果是第一个数据源，这个latch可以设为已完成状态，无需等待）。

public class ControlledSequentialSourceFunction<T> implements SourceFunction<T> {
    private final List<T> data;
    private final CountDownLatch startLatch;
    private volatile boolean isRunning = true;

    // 第一个数据源用new CountDownLatch(0)（无需等待），第二个传入等待第一个完成的latch
    public ControlledSequentialSourceFunction(List<T> data, CountDownLatch startLatch) {
        this.data = data;
        this.startLatch = startLatch;
    }

    @Override
    public void run(SourceContext<T> ctx) throws Exception {
        // 等待前置信号（如果有的话）
        startLatch.await();

        synchronized (ctx.getCheckpointLock()) {
            for (T element : data) {
                if (!isRunning) break;
                ctx.collect(element);
                // 可选：模拟真实场景的事件延迟，按需调整
                Thread.sleep(10);
            }
        }
    }

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

2. 在集成测试中协调两个数据源的执行顺序 #

在测试代码里，我们需要创建同步工具关联两个数据源：第一个数据源完成后触发第二个数据源启动，同时按照官方集成测试流程把作业提交到MiniCluster运行。

@Test
public void testCoFlatMapJobWithSequentialSources() throws Exception {
    // 1. 准备测试数据
    List<String> source1Data = Arrays.asList("data1-1", "data1-2", "data1-3");
    List<String> source2Data = Arrays.asList("data2-1", "data2-2", "data2-3");

    // 2. 初始化同步工具：source1完成后触发source2启动
    CountDownLatch source1CompletionLatch = new CountDownLatch(1);

    // 3. 创建带控制的数据源
    SourceFunction<String> source1 = new ControlledSequentialSourceFunction<>(
        source1Data,
        new CountDownLatch(0) // 第一个数据源无需等待，直接启动
    ) {
        // 重写run方法，发射完所有数据后触发latch通知source2
        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            super.run(ctx);
            source1CompletionLatch.countDown();
        }
    };

    SourceFunction<String> source2 = new ControlledSequentialSourceFunction<>(
        source2Data,
        source1CompletionLatch // 等待source1完成的信号
    );

    // 4. 构建并提交Flink作业到MiniCluster（集成测试标准流程）
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 根据你的作业实际配置调整并行度、检查点等参数
    env.setParallelism(1);

    DataStream<String> stream1 = env.addSource(source1);
    DataStream<String> stream2 = env.addSource(source2);

    // 替换成你实际的CoFlatMap业务逻辑
    DataStream<String> result = stream1.connect(stream2)
        .flatMap(new YourCoFlatMapFunction());

    // 5. 收集测试结果并验证
    List<String> collectedResult = Collections.synchronizedList(new ArrayList<>());
    result.addSink(new SinkFunction<String>() {
        @Override
        public void invoke(String value, Context context) {
            collectedResult.add(value);
        }
    });

    // 执行作业
    env.execute("Sequential Source Integration Test");

    // 6. 根据业务逻辑编写断言验证结果
    // 示例：确保source1的处理结果全部出现在source2结果之前
    int lastSource1Index = collectedResult.lastIndexOf("processed-data1-3");
    int firstSource2Index = collectedResult.indexOf("processed-data2-1");
    assertTrue(lastSource1Index < firstSource2Index);
}

关键注意事项 #

• 并行度适配：如果数据源设置了并行度>1，需要调整同步机制（比如用CyclicBarrier或多个CountDownLatch），确保所有并行的source1实例都完成后再启动source2。
• 环境兼容性：该方案完全基于Flink原生API，适配官方文档推荐的MiniCluster集成测试场景，不会和现有测试流程冲突。
• 资源清理：测试结束后要确保latch和SourceFunction的资源被正确释放，避免影响后续测试用例。

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