java

13.2 Flink DataStream 测试

Flink DataStream 测试体系：MiniCluster 资源管理与 execute() 阻塞模式、KeyedOneInputStreamOperatorTestHarness 进行 ProcessFunction/Timer 精细测试、CollectSink 收集结果、Watermark 手动推进测试，以及 eventd v3 窗口聚合的确定性测试。

Leeting Yan 2026-11-01 6 分钟阅读 2618 字

Flink 测试分层

层级	工具	范围	速度
算子测试	TestHarness	单个算子	< 1s
作业测试	MiniCluster	完整作业	3-10s
集成测试	Testcontainers	+ Kafka/Redis	30-60s

eventd v3 的测试覆盖全部三层。

MiniCluster 基础测试

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

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

    cluster.start();
    try {
        // 关键一步：把 getExecutionEnvironment() 绑到这个 MiniCluster 上。
        // 少了它，env 会另建一个 LocalStreamEnvironment，作业根本没提交到 cluster，
        // cluster 只是个空跑的背景进程。
        TestStreamEnvironment.setAsContext(cluster, 2);

        StreamExecutionEnvironment env = StreamExecutionEnvironment
            .getExecutionEnvironment();
        env.setParallelism(2);

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

        DataStream<Integer> result = stream
            .map(x -> x * x)
            .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(25));  // 5 * 5
        assertTrue(output.contains(100)); // 10 * 10
    } finally {
        TestStreamEnvironment.unsetAsContext();
        cluster.close();
    }
}

setAsContext 是全局静态状态，必须在 finally 里 unsetAsContext()，否则同一个 JVM 里后续测试会继续往这个已关闭的集群提交作业。JUnit 5 项目更省事的写法是用 MiniClusterExtension，它在 @BeforeAll/@AfterAll 里做了同样的事：

class AnalyticsJobIT {
    // 注意构造参数是 MiniClusterResourceConfiguration，
    // 不是前面手写 MiniCluster 时用的 MiniClusterConfiguration
    @RegisterExtension
    static final MiniClusterExtension MINI_CLUSTER =
        new MiniClusterExtension(new MiniClusterResourceConfiguration.Builder()
            .setNumberTaskManagers(1)
            .setNumberSlotsPerTaskManager(2)
            .build());

    @Test
    void testJob() throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        // ... 建流、env.execute()
    }
}

TestHarness：精细算子测试

KeyedOneInputStreamOperatorTestHarness 测试 KeyedProcessFunction：

// 这个测试只关心「谁 + 什么时候」，用一个只含必要字段的 record 当夹具，
// 不必构造 v3 那个 12 个组件的 Event
record UserEvent(String userId, long timestamp) {}

@Test
void testTimerFunction() throws Exception {
    KeyedOneInputStreamOperatorTestHarness<String, UserEvent, String> harness =
        new KeyedOneInputStreamOperatorTestHarness<>(
            new KeyedProcessOperator<>(new MyTimerFunction()),
            UserEvent::userId,   // key selector
            Types.STRING
        );

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

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

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

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

    // extractOutputValues() 直接给出输出值，省掉手工拆 StreamRecord
    List<String> results = harness.extractOutputValues();
    assertEquals(1, results.size());
    assertTrue(results.get(0).contains("alice"));

    harness.close();
}

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

Watermark 推进测试

窗口测试的关键是控制 Watermark 何时到达：

@Test
void testTumblingWindow() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment
        .createLocalEnvironment(1);
    // 不要写 env.setStreamTimeCharacteristic(...)：1.12 起已废弃且不生效，
    // 时间语义由下面 assignTimestampsAndWatermarks 的 WatermarkStrategy 决定

    // 只要「时间戳 + 数值」两个字段，用局部 record 即可
    record TimedValue(long timestamp, double amount) {}

    // 使用带时间戳的测试数据
    DataStream<TimedValue> stream = env.fromCollection(Arrays.asList(
        new TimedValue(1000, 100.0),   // timestamp = 1s
        new TimedValue(2000, 200.0),   // timestamp = 2s
        new TimedValue(3000, 300.0),   // timestamp = 3s
        new TimedValue(6000, 600.0)    // timestamp = 6s → Watermark = 6s - 1ms
    )).assignTimestampsAndWatermarks(
        WatermarkStrategy.<TimedValue>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.timestamp())
    );

    // 5 秒滚动窗口
    DataStream<Double> result = stream
        .map(TimedValue::amount)
        .windowAll(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
        .reduce((a, b) -> a + b);

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

    env.execute("Window Test");

    // 有界源读完后会补发 Watermark = Long.MAX_VALUE，
    // 所以两个窗口都会触发，而不是只到 6s
    List<Double> output = CollectSink.values();
    assertEquals(2, output.size());
    // 窗口 [0, 5000): 100 + 200 + 300 = 600
    assertEquals(600.0, output.get(0), 0.001);
    // 窗口 [5000, 10000): 只剩 600 那一条
    assertEquals(600.0, output.get(1), 0.001);
}

这一点很容易写错：注释里写「execute() 结束时 Watermark 只到 6s，所以第二个窗口不触发」是错的。fromCollection 是有界源，读完后 BoundedOneInput.endInput() 会让 watermark 生成器发最大值，剩余窗口照样触发。要测「窗口不触发」，不能靠调用顺序，必须用 TestHarness 精确控制 Watermark。

使用 TestHarness 可以更精确控制 Watermark：

// v3 的 Event 是 12 个组件的 record；小工厂只写关心的字段，
// 其余取默认值（version=1、source=eventd-v2、status=0）
private static Event event(String id, String userId, double amount, long ts) {
    return new Event(id, "ORDER", 1, "eventd-v2", ts, userId, amount, "BTC",
            0, "127.0.0.1", "web", Map.of());
}

// 窗口求和：8.3 的 SumAggregate 聚合 Integer，这里聚合 Event 的 amount
public class SumAggregate
        implements AggregateFunction<Event, Double, Double> {
    @Override
    public Double createAccumulator() { return 0.0; }

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

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

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

@Test
void testWindowWithWatermark() throws Exception {
    // WindowOperator 的公开构造函数有 9 个参数（TypeSerializer、StateDescriptor、
    // InternalWindowFunction、Trigger、OutputTag…），手写几乎必错。
    // 用 WindowOperatorBuilder：它和 WindowedStream.apply() 内部是同一套装配逻辑
    WindowOperatorBuilder<Event, String, TimeWindow> builder =
        new WindowOperatorBuilder<>(
            TumblingEventTimeWindows.of(Duration.ofSeconds(5)),
            EventTimeTrigger.create(),
            new ExecutionConfig(),
            TypeInformation.of(Event.class),
            Event::userId,
            Types.STRING);

    WindowOperator<String, Event, ?, Double, TimeWindow> operator =
        builder.aggregate(
            new SumAggregate(),   // AggregateFunction<Event, Double, Double>
            (WindowFunction<Double, Double, String, TimeWindow>)
                (key, window, input, out) -> out.collect(input.iterator().next()),
            // 第三个参数是累加器 ACC 的类型信息，不是输出类型
            Types.DOUBLE);

    KeyedOneInputStreamOperatorTestHarness<String, Event, Double> harness =
        new KeyedOneInputStreamOperatorTestHarness<>(
            operator, Event::userId, Types.STRING);

    try {
        harness.setup();
        harness.open();

        harness.processElement(new StreamRecord<>(event("e1", "alice", 100.0, 1000), 1000));
        harness.processElement(new StreamRecord<>(event("e2", "alice", 200.0, 2000), 2000));

        // 手动推进 Watermark 到 5000 → 触发窗口 [0, 5000)
        harness.processWatermark(new Watermark(5000));

        List<Double> results = harness.extractOutputValues();

        assertEquals(1, results.size());
        assertEquals(300.0, results.get(0), 0.001);  // 100 + 200
    } finally {
        harness.close();
    }
}

KeyedOneInputStreamOperatorTestHarness 的字节码在 flink-streaming-java 的 tests 分类器里，由 flink-test-utils（test scope）传递到测试 classpath，不用单独声明。harness.getOutput() 里既有 StreamRecord 也有 Watermark 等对象，手工遍历必须过滤；能用 extractOutputValues() 就别自己拆。

CollectSink

public class CollectSink<T> implements SinkFunction<T> {
    // 必须是 static final：SinkFunction 在部署前会被序列化，
    // 执行的是反序列化出来的新对象，实例字段拿到的是副本
    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;
    }
}

为什么非得是 static？把集合作为构造参数传进去（new CollectSink<>(output)）看起来更干净，但收不到数据。Flink 提交作业时会把算子函数序列化进 JobGraph，TaskManager 侧反序列化出一个新对象；那个副本里的 output 字段指向反序列化出来的列表，测试线程手里的是另一个。实测本地执行器（并行度 1）也是这样：

实例字段 CollectSink → 测试线程读到的 size = 0
静态字段 CollectSink → 测试线程读到的 size = 3

静态字段不参与 Java 序列化，反序列化后的副本和测试线程看到的是同一个 JVM 里的同一份静态变量，所以在本地执行器和同进程 MiniCluster 里都能用。

代价是静态变量跨测试共享，必须在每个测试前清空：

@BeforeEach
void reset() {
    CollectSink.VALUES.clear();
}

也不要把它当集群测试的手段：跨进程部署时静态字段各进程一份，测试 JVM 依然读不到。需要跨进程断言就用 Testcontainers 把结果写到外部系统再查。

eventd v3 窗口聚合测试

@Test
void testOrderCountAggregate() throws Exception {
    // event(...) 工厂沿用上一段测试里的定义
    StreamExecutionEnvironment env = StreamExecutionEnvironment
        .createLocalEnvironment(1);

    // 用对齐到窗口边界的时间戳，而不是 now()：
    // 5 秒窗口按 epoch 取整，now() 落在窗口中间时，
    // now()+6000 可能与前三条不在预期的两个窗口里
    long base = 1_700_000_000_000L;  // 能被 5000 整除

    List<Event> events = Arrays.asList(
        event("e1", "alice", 100, base),
        event("e2", "alice", 200, base + 1000),
        event("e3", "alice", 300, base + 2000),
        // 这条把 Watermark 推到 base+5999，触发 [base, base+5000) 窗口
        event("e4", "alice", 400, base + 6000)
    );

    DataStream<Event> stream = env.fromCollection(events)
        .assignTimestampsAndWatermarks(
            WatermarkStrategy.<Event>forMonotonousTimestamps()
                .withTimestampAssigner((e, ts) -> e.eventTime())
        );

    stream.keyBy(Event::userId)
        .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
        .aggregate(new OrderCountAggregate())
        .addSink(new CollectSink<>());

    env.execute("Aggregate Test");

    // 有界源结束会补发 Long.MAX_VALUE，第二个窗口也会触发：
    // [base, base+5000) 3 条，[base+5000, base+10000) 1 条
    List<OrderCountResult> output = CollectSink.values();
    assertEquals(2, output.size());
    assertEquals(3L, output.get(0).count());
    assertEquals(1L, output.get(1).count());
}

测试最佳实践

实践	原因
先用 `createLocalEnvironment`	最快，适合大部分场景
需要 Checkpoint 时用 `MiniCluster`	真实集群环境
并行度设为 1	避免输出乱序，且 `CollectSink` 的写回才有确定顺序
使用 `forMonotonousTimestamps()`	测试数据通常按顺序
时间戳对齐窗口边界	用 `now()` 会让事件跨在窗口分界上
`CollectSink` 用静态集合	函数会被序列化，实例字段拿到的是副本
断言前先排序	多并行度或多次触发时输出顺序不保证

小结

工具	用途
`createLocalEnvironment`	快速作业测试
`MiniCluster`	集群模拟
`TestHarness`	精细算子测试
`CollectSink`	结果收集

下一节 13.3 端到端测试：Testcontainers + 全链路集成。

继续阅读

探索更多技术文章

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

全部文章返回首页