Fixed timer deletion bug in CoProcessFunctionTimers example (Chapter 6)

streaming-with-flink · Jul 24, 2019 · b448e74 · b448e74
1 parent 609d13e
commit b448e74
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/src/main/scala/io/github/streamingwithflink/chapter6/CoProcessFunctionTimers.scala b/src/main/scala/io/github/streamingwithflink/chapter6/CoProcessFunctionTimers.scala
@@ -84,7 +84,7 @@ class ReadingFilter
     val curTimerTimestamp = disableTimer.value()
     if (timerTimestamp > curTimerTimestamp) {
       // remove current timer and register new timer
-      ctx.timerService().deleteEventTimeTimer(curTimerTimestamp)
+      ctx.timerService().deleteProcessingTimeTimer(curTimerTimestamp)
       ctx.timerService().registerProcessingTimeTimer(timerTimestamp)
       disableTimer.update(timerTimestamp)
     }