如何使用触发器和驱逐器测试Flink全局窗口

Question

我有一个使用Flink Global Window的管道，它具有基于事件时间（从到达元素的时间戳）的自定义触发器和从窗口中删除不必要元素并将其传递给ProcessFunction的Evictor，

像这样的东西：

 public SingleOutputStreamOperator<Results> processElements(DataStream<Elements> inputStream) {
 return inputStream
                .keyBy(Elements::getId)
                .window(GlobalWindows.create())
                .trigger(new CustomTrigger())
                .evictor(new CustomEvictor())
                .process(new MyWindowProcessFunction())
                .name("Process")
                .uid("process-elements")
                .returns(Results.class);    
}

    public void executePipelineFlow(StreamExecutionEnvironment env) throws Exception {
        DataStream<Elements> inputStream = getInputStream(env);
        DataStream<Results> processedInput = processElements(inputStream);
        applySink(processedInput);
}

我知道我可以用提供水印操作的TestHarness测试我的WindowProcessFunction，但是我需要测试整个流程，触发退出器ProcessFunction。

我还尝试了使用Thread.sleep()的某种定时SourceFunction，但是我的管道在事件时间内工作，如果我在测试流中有1000个元素，这将不起作用(因为测试将需要几个小时)。

我的问题是，如何对整个< code>processElements方法进行单元测试？

我找不到我的案例的任何测试示例。

谢谢

Answer 1

作为一个例子，您可以看看Flink培训中窗口练习的端到端集成测试是如何实现的。这个练习没有使用GlobalWindows或自定义触发等，但是您可以使用这个整体方法来测试任何管道。

这种方法可能不太理想的一点是它如何处理水印。被测试的应用程序使用默认的周期性水印策略，其中水印每200毫秒生成一次。由于测试不会运行那么长时间，实际生成的唯一水印是在每个输入受限的作业结束时出现的水印。这是可行的，但与生产中会发生的情况不太一样。(这就是你考虑让测试源在事件之间Hibernate的原因吗？)

顺便说一下，Flink培训报告中的这些测试比通常需要的稍微复杂一些，因为这些测试用于覆盖练习和解决方案的Java和Scala实现。