apache-flink - 使用 MiniCluster 测试 flink 作业以使用处理时间触发计时器

Question

在使用MiniClusterWithClientResource测试 flink 作业时，有什么方法可以控制触发计时器的处理时间？

我能够使用测试工具在单元测试中测试 KeyedCoProcessFunction 即 processElement()... 触发计时器回调即 onTimer()...的两种方法并控制处理时间，即：

//通过直接提前算子的处理时间触发处理时间计时器 testHarness.setProcessingTime(300000)

因此。我可以在指定的时间触发定时器

但是，我现在需要的是使用 minicluster MiniClusterWithClientResource在端到端的 flink 作业测试中触发计时器

val flinkCluster = new MiniClusterWithClientResource... 并且能够提前处理时间来触发 onTimer 方法

Answer 1

发送完所有消息后，在 SourceFunction 类中添加 Thread.sleep(1000) 一秒即可解决问题。

class MySourceFunction() extends RichParallelSourceFunction[]{
...

//is a one-time Delay after all messages have been sent
Thread.sleep(1000)
}