java

8.3 测试：MiniCluster 与测试工具

Flink 作业的确定性测试方法：StreamExecutionEnvironment.createLocalEnvironment() 的测试模式、CollectSink 收集结果、Watermark 推进测试、TestStreamEnvironment 与 TestHarness、以及基于 4.1 的 POJO Event 的端到端可运行测试示例。

Leeting Yan 5 分钟阅读 2448 字

流处理测试的挑战

批处理测试简单：输入固定数据集 → 运行作业 → 断言输出。流处理测试更复杂：

  • 数据是持续到达的，没有明确的「结束」
  • 时间是关键变量（Watermark、窗口触发）
  • 乱序数据的行为需要验证
  • 作业可能永不结束（长期运行的流作业）

Flink 提供了多种测试工具来应对这些挑战。

本地环境测试

最简单的测试方式：本地环境 + 有限数据 + 同步执行。

StreamExecutionEnvironment.createLocalEnvironment()

@Test
void testWordCount() throws Exception {
    // 创建本地环境（不是集群）
    StreamExecutionEnvironment env = StreamExecutionEnvironment
        .createLocalEnvironment(1);  // 并行度 1

    // 有限数据源（使用 fromCollection）
    DataStream<String> text = env.fromCollection(Arrays.asList(
        "hello world",
        "hello flink",
        "hello java"
    ));

    DataStream<Tuple2<String, Integer>> counts = text
        .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            for (String word : value.split(" ")) {
                out.collect(Tuple2.of(word, 1));
            }
        })
        .returns(Types.TUPLE(Types.STRING, Types.INT))
        .keyBy(v -> v.f0)
        .sum(1);

    // 收集结果
    CollectSink.VALUES.clear();
    counts.addSink(new CollectSink<>());

    // 同步执行（阻塞直到完成）
    env.execute("WordCount Test");

    // 断言
    List<Tuple2<String, Integer>> result = CollectSink.values();
    assertTrue(result.contains(Tuple2.of("hello", 3)));
    assertTrue(result.contains(Tuple2.of("world", 1)));
    assertTrue(result.contains(Tuple2.of("flink", 1)));
    assertTrue(result.contains(Tuple2.of("java", 1)));
}

注意这里没有 setStreamTimeCharacteristic(EventTime)。Flink 1.12 起时间语义由 WatermarkStrategy 决定，这个方法已废弃且不会改变窗口的时间语义；测试里需要事件时间时，直接给数据分配时间戳与 Watermark。

CollectSink

public class CollectSink<T> implements SinkFunction<T> {
    // static 不是风格问题，是能否收到数据的问题（见下）
    static final List<Object> VALUES =
        Collections.synchronizedList(new ArrayList<>());

    @Override
    public void invoke(T value, Context context) {
        VALUES.add(value);
    }

    @SuppressWarnings("unchecked")
    static <T> List<T> values() {
        return (List<T>) (List<?>) VALUES;
    }
}

三个细节决定这个 Sink 能不能用：

  • 集合必须是 static。把 list 当构造参数传进去（new CollectSink<>(result)）看起来更干净，但断言时 result 永远是空的：算子函数在提交前会被序列化进 JobGraph，执行的是反序列化出来的新对象，它的字段指向另一个 list。实测本地执行器（并行度 1）也是这个结果。静态字段不参与序列化，同 JVM 内测试线程与算子副本才能看到同一份数据。
  • invoke 会被多个子任务并发调用，list 要同步；并行度 > 1 时还要按内容排序或转成集合再断言，不能依赖顺序。
  • 静态集合是 JVM 级的，必须每个测试前 clear()，否则测试之间互相污染。跨进程部署时各进程一份静态变量，测试 JVM 依然看不到——那种场景要用 TestHarness 或把结果写回可查询的存储。

13.2 会给出实测对比和更完整的取舍。

Watermark 推进测试

测试窗口操作需要控制 Watermark 的推进：

@Test
void testWindowAggregation() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment
        .createLocalEnvironment(1);

    // 本测试只关心「时间戳 + 数值」，用一个局部 record，避免和 E2E 测试里的 Event 混淆
    record TimedValue(long timestamp, int value) {}

    // 创建带时间戳和 Watermark 的测试数据
    DataStream<TimedValue> stream = env.fromCollection(Arrays.asList(
        new TimedValue(1000, 1),   // timestamp 1s
        new TimedValue(2000, 2),   // timestamp 2s
        new TimedValue(3000, 3),   // timestamp 3s
        new TimedValue(6500, 4)    // timestamp 6.5s，Watermark = 6.5 - 1 = 5.5s
    )).assignTimestampsAndWatermarks(
        WatermarkStrategy.<TimedValue>forBoundedOutOfOrderness(Duration.ofSeconds(1))
            .withTimestampAssigner((event, ts) -> event.timestamp())
    );

    DataStream<Integer> result = stream
        .map(TimedValue::value)
        .windowAll(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
        .aggregate(new SumAggregate());

    CollectSink.VALUES.clear();
    result.addSink(new CollectSink<>());

    env.execute("Window Test");

    // 有界源结束时，Flink 会发出 Long.MAX_VALUE 的 Watermark，所有窗口都会触发
    // [0, 5000)    → 1 + 2 + 3 = 6
    // [5000, 10000) → 4
    List<Integer> output = CollectSink.values();
    assertEquals(2, output.size());
    assertEquals(List.of(4, 6), output.stream().sorted().toList());
}

// 窗口求和的聚合函数：这里是 Integer，13.2 的版本换成 Event/Double
public class SumAggregate
        implements AggregateFunction<Integer, Integer, Integer> {
    @Override
    public Integer createAccumulator() { return 0; }

    @Override
    public Integer add(Integer value, Integer acc) { return acc + value; }

    @Override
    public Integer getResult(Integer acc) { return acc; }

    @Override
    public Integer merge(Integer a, Integer b) { return a + b; }
}

// E2E 用的是 Event → Double 的版本，不能复用上面的 Integer 版
public class AmountSumAggregate
        implements AggregateFunction<Event, Double, Double> {
    @Override
    public Double createAccumulator() { return 0.0; }

    @Override
    public Double add(Event event, Double acc) { return acc + event.getAmount(); }

    @Override
    public Double getResult(Double acc) { return acc; }

    @Override
    public Double merge(Double a, Double b) { return a + b; }
}

MiniCluster 测试

MiniCluster 比 createLocalEnvironment 更接近真实集群：

@Test
void testWithMiniCluster() throws Exception {
    Configuration config = new Configuration();
    config.setInteger(RestOptions.PORT, 0);  // 随机端口

    MiniCluster miniCluster = new MiniCluster(
        new MiniClusterConfiguration.Builder()
            .setNumTaskManagers(1)
            .setNumSlotsPerTaskManager(2)
            .setConfiguration(config)
            .build());

    miniCluster.start();

    try {
        // 关键一步：把 MiniCluster 设为当前上下文。
        // 只 start() 再 getExecutionEnvironment() 并不会提交到这个集群，
        // getExecutionEnvironment() 仍会创建一个独立的本地环境。
        TestStreamEnvironment.setAsContext(miniCluster, 2);

        StreamExecutionEnvironment env = StreamExecutionEnvironment
            .getExecutionEnvironment();
        env.setParallelism(2);  // 使用 2 个 slot

        DataStream<Integer> input = env.fromCollection(
            Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

        DataStream<Integer> result = input
            .map(x -> x * 2)
            .setParallelism(2);

        CollectSink.VALUES.clear();
        result.addSink(new CollectSink<>());

        env.execute("MiniCluster Test");

        List<Integer> output = CollectSink.values();
        assertEquals(10, output.size());
        assertTrue(output.contains(2));
        assertTrue(output.contains(20));
    } finally {
        TestStreamEnvironment.unsetAsContext();
        miniCluster.close();
    }
}

TestStreamEnvironment 位于 org.apache.flink.streaming.util，由 flink-test-utils（test scope）提供；MiniCluster 与 MiniClusterConfiguration 在 flink-runtime。JUnit 5 项目更常用 org.apache.flink.test.junit5.MiniClusterExtension，它会把启动/关闭和上下文设置一起管好，写法见 13.2 Flink DataStream 测试 。

MiniCluster 测试的好处：

  • 真实网络通信（TM 之间）
  • Checkpoint、Savepoint 功能可用
  • 更接近生产环境的执行模式

ProcessFunction 测试

KeyedProcessFunction 需要更精细的测试——手动传入状态、注册 Timer：

// 测试夹具独立定义，避免和业务 Event 的字段/构造器纠缠
record UserEvent(String userId, long timestamp) {}

@Test
void testTimerFunction() throws Exception {
    // KeyedProcessFunction 的测试工具
    KeyedOneInputStreamOperatorTestHarness<String, UserEvent, String> harness =
        new KeyedOneInputStreamOperatorTestHarness<>(
            new KeyedProcessOperator<>(new MyTimerFunction()),
            UserEvent::userId,
            Types.STRING
        );

    harness.setup();
    harness.open();

    // 处理时间设为 0
    harness.setProcessingTime(0);

    // 发送事件
    harness.processElement(new UserEvent("alice", 1000L), 0);
    harness.processElement(new UserEvent("alice", 2000L), 0);

    // 推进处理时间到 1 分钟后（触发 Timer）
    harness.setProcessingTime(60_000);

    // extractOutputValues() 直接给出输出值列表，省掉手工拆 StreamRecord
    List<String> results = harness.extractOutputValues();
    assertEquals(1, results.size());

    harness.close();
}

要让这个测试成立，MyTimerFunction 必须真的在 processElement 里注册处理时间 Timer（例如 ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() + 60_000)），并在 onTimer 中 out.collect(...)。只调用 setProcessingTime、却没有对应 Timer 时，extractOutputValues() 是空列表——这是测试写错时最常见的现象。

getOutput() 返回的是 ConcurrentLinkedQueue<Object>，里面混着 StreamRecord、Watermark 等对象，需要自己 instanceof 过滤；能用 extractOutputValues() 就别手工遍历。

端到端测试

第 8 章仍在用 4.1 定义的 POJO Event（v3 的 record 模型到 9.1 才引入），端到端测试因此用 POJO 的 4 参构造与 getter；9.1 之后 record 版本的测试见 13.2 。

// 用「第 n 秒」构造时间戳，让窗口边界一眼可见
private static long instant(long seconds) {
    return seconds * 1000;
}

@Test
void testEventdE2E() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment
        .createLocalEnvironment(2);

    // 1. 模拟事件源
    DataStream<Event> events = env.fromCollection(Arrays.asList(
        new Event("e1", "ORDER", 100.0, instant(0)),
        new Event("e2", "ORDER", 200.0, instant(1)),
        new Event("e3", "ORDER", 300.0, instant(2)),
        new Event("e4", "REFUND", 50.0, instant(65))  // 跨过窗口边界：落在第二个窗口
    )).assignTimestampsAndWatermarks(
        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner((e, ts) -> e.getEventTime())
    );

    // 2. 清洗（filter）
    DataStream<Event> cleaned = events.filter(e -> e.getAmount() > 0);

    // 3. 窗口聚合
    DataStream<Double> metrics = cleaned
        .keyBy(Event::getType)
        .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
        .aggregate(new AmountSumAggregate());

    // 4. 收集结果
    CollectSink.VALUES.clear();
    metrics.addSink(new CollectSink<>());

    env.execute("E2E Test");

    // 断言：两个窗口 [0,60) 和 [60,120)，按金额排序后比较，不依赖输出顺序
    // （sink 并行度 2，两个子任务的写入交错，顺序本来就不确定）
    List<Double> totals = CollectSink.<Double>values().stream()
        .sorted()
        .toList();
    assertEquals(List.of(50.0, 600.0), totals);  // [60,120) 只有 50，[0,60) 是 100+200+300
}

测试最佳实践

实践原因
使用有限数据集让测试可结束
使用 fromCollection替代外部 Source（Kafka）
使用 CollectSink收集结果做断言
控制 Watermark窗口单元测试的关键
并行度设为 1 或固定值避免结果乱序
使用 MiniCluster 做集成测试模拟真实环境

小结

工具用途
createLocalEnvironment()简单单元测试
MiniCluster集群环境集成测试
CollectSink收集结果
TestHarnessProcessFunction 精细测试
setProcessingTime()推进处理时间触发 Timer
assignTimestampsAndWatermarks()测试窗口/Watermark 行为

第 8 章结束。下一节进入 eventd v3 实现：9.1 eventd v3 架构 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南