Flink的CoProcessFunction不会触发onTimer

Question

我尝试聚合这样的两个流

val joinedStream = finishResultStream.keyBy(_.searchId)
  .connect(startResultStream.keyBy(_.searchId))
  .process(new SomeCoProcessFunction)

然后在SomeCoProcessFunction类中对它们进行研究

class SomeCoProcessFunction extends CoProcessFunction[SearchFinished, SearchCreated, SearchAggregated] {

   override def processElement1(finished: SearchFinished, ctx: CoProcessFunction[SearchFinished, SearchCreated, SearchAggregated]#Context, out: Collector[SearchAggregated]): Unit = { 

       // aggregating some "finished" data ...

   }

   override def processElement2(created: SearchCreated, ctx: CoProcessFunction[SearchFinished, SearchCreated, SearchAggregated]#Context, out: Collector[SearchAggregated]): Unit = {

       val timerService = ctx.timerService()
       timerService.registerEventTimeTimer(System.currentTimeMillis + 5000)

       // aggregating some "created" data ...
   }

   override def onTimer(timestamp: Long, ctx: CoProcessFunction[SearchFinished, SearchCreated, SearchAggregated]#OnTimerContext, out: Collector[SearchAggregated]): Unit = {

       val watermark: Long = ctx.timerService().currentWatermark()
       println(s"watermark!!!! $watermark")

       // clean up the state

   }

我想要的是在一定时间（5000毫秒）后清理状态，这就是必须使用onTimer。但既然它永远不会被解雇，我有点问自己，我在这里做错了什么？

提前感谢任何提示。

更新：

解决方案是设置timeService（tnx到fabian-hueske和Beckham）：

timerService.registerProcessingTimeTimer(timerService.currentProcessingTime() + 5000)

我仍然没有真正弄清楚timerService.registerEventTimeTimer做了什么，水印ctx.timerService().currentWatermark()显示总是-9223372036854775808现在问题在EventTimer注册之前多久。

Answer 1

问题是您正在使用处理时间戳（timerService.registerEventTimeTimer）注册事件时间计时器（System.currentTimeMillis + 5000）。

System.currentTimeMillis返回当前机器时间，但事件时间不是基于机器时间，而是基于水印计算的时间。

您应该注册处理定时器或使用事件时间戳记注册事件时间定时器。您可以从Context对象获取当前水印的时间戳或当前记录的时间戳，该对象作为参数传递给processElement1()和processElement2()。

Answer 2

我发现你使用的System.currentTimeMillis可能与你的Flink作业使用的TimeCharacteristic（事件时间，处理时间，摄取时间）不同。

尝试获取事件ctx.timestamp()的时间戳，然后在其上添加5000毫秒。