Fixed condition in InputDStream isTimeValid.

2013-02-21 15:22:26 -08:00 · 2013-02-21 15:22:26 -08:00 · 208edaac1b
parent 972fe7714f
commit 208edaac1b
1 changed files with 1 additions and 1 deletions
--- a/streaming/src/main/scala/spark/streaming/dstream/InputDStream.scala
+++ b/streaming/src/main/scala/spark/streaming/dstream/InputDStream.scala
@ -29,7 +29,7 @@ abstract class InputDStream[T: ClassManifest] (@transient ssc_ : StreamingContex
      false // Time not valid
    } else {
      // Time is valid, but check it it is more than lastValidTime
-      if (lastValidTime == null || lastValidTime <= time) {
+      if (lastValidTime != null && time < lastValidTime) {
        logWarning("isTimeValid called with " + time + " where as last valid time is " + lastValidTime)
      }
      lastValidTime = time