我试着像那样把两条溪流聚合起来

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（对fabian hueske和beckham都是tnx）： timerService.registerProcessingTimeTimer(timerService.currentProcessingTime() + 5000) 我还是没弄清楚是什么 timerService.registerEventTimeTimer 是的，水印 ctx.timerService().currentWatermark() 始终显示 -9223372036854775808 现在不管eventtimer注册前多久。