Improve ProcessFunctionTimers example (Chapter 6)

streaming-with-flink · Jul 24, 2019 · c188681 · c188681
1 parent b448e74
commit c188681
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/src/main/scala/io/github/streamingwithflink/chapter6/ProcessFunctionTimers.scala b/src/main/scala/io/github/streamingwithflink/chapter6/ProcessFunctionTimers.scala
@@ -64,8 +64,12 @@ class TempIncreaseAlertFunction
     // update last temperature
     lastTemp.update(r.temperature)
 
-    val curTimerTimestamp = currentTimer.value();
-    if (prevTemp == 0.0 || r.temperature < prevTemp) {
+    val curTimerTimestamp = currentTimer.value()
+    if (prevTemp == 0.0) {
+      // first sensor reading for this key.
+      // we cannot compare it with a previous value.
+    }
+    else if (r.temperature < prevTemp) {
       // temperature decreased. Delete current timer.
       ctx.timerService().deleteProcessingTimeTimer(curTimerTimestamp)
       currentTimer.clear()