未触发具有事件时间特性的TumblingProcessingTimeWindows处理

Question

更具体地说，需要计算的加密数据流的计数为7秒。

1. 1秒滚动窗口
2. 可提前1秒计算7秒的滑动窗口
3. 每隔1秒输出全部计数的windowall

我无法对其进行集成测试（即，类似于单元测试，但端到端测试），因为输入有假事件时间，这不会触发

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val oneDayCounts = data
  .map(t => (t.key1, t.key2, 1L, t.timestampMs))
  .keyBy(0, 1)
  .timeWindow(Time.seconds(1))
  .sum(2)

val sevenDayCounts = oneDayCounts
  .keyBy(0,1)
  .timeWindow(Time.seconds(3), Time.seconds(1))
  .sum(2)

// single reducer
sevenDayCounts
  .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
  .process(...)

EventTimeTrigger可以为求和计算激发，但以下TumblingProcessingTimeWindow无法触发。我在IT测试代码中有一个30s的线程.睡眠，但在30s之后仍然没有触发

Answer 1

一般来说，编写有意义的测试来处理时间窗口是一个挑战，因为它们本质上是不确定的。这是为什么通常首选事件时间窗口的一个原因。

在正确的地方睡觉也很难达到预期的效果。但是，保持作业运行足够长的时间以使处理时间窗口触发的一种方法是使用包含Hibernate的自定义源。具有有限源的Flink流作业在输入耗尽后自行关闭。通过管道发送值为MAX_WATERMARK的最后一个水印，这将触发所有事件时间窗口，但处理时间窗口只有在指定时间到达时仍在运行时才会触发。

请参阅此答案以获得一个解决此问题的黑客示例。