From 208edaac1bdb4b0520e5fb204f7bc96079bfa3ed Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Thu, 21 Feb 2013 15:22:26 -0800
Subject: Fixed condition in InputDStream isTimeValid.

---
 streaming/src/main/scala/spark/streaming/dstream/InputDStream.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'streaming')

diff --git a/streaming/src/main/scala/spark/streaming/dstream/InputDStream.scala b/streaming/src/main/scala/spark/streaming/dstream/InputDStream.scala
index a4db44a608..3c5d43a609 100644
--- 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
-- 
cgit v1.2.3