MinTimeFirehose: Correcter logic

realtime/src/main/java/com/metamx/druid/realtime/MinTimeFirehose.java

@@ -4,12 +4,19 @@ import com.metamx.druid.input.InputRow;
 import org.joda.time.DateTime;
 
 import java.io.IOException;
+import java.util.NoSuchElementException;
 
+/**
+ * Provides a view on a firehose that only returns rows at or after a certain minimum timestamp.
+ * Not thread-safe.
+ */
 public class MinTimeFirehose implements Firehose
 {
   private final Firehose firehose;
   private final DateTime minTime;
 
+  private InputRow savedInputRow = null;
+
   public MinTimeFirehose(Firehose firehose, DateTime minTime)
   {
     this.firehose = firehose;
@@ -19,18 +26,38 @@ public class MinTimeFirehose implements Firehose
   @Override
   public boolean hasMore()
   {
-    return firehose.hasMore();
+    if (savedInputRow != null) {
+      return true;
+    }
+
+    while (firehose.hasMore()) {
+      final InputRow row = firehose.nextRow();
+      if (acceptable(row)) {
+        savedInputRow = row;
+        return true;
+      }
+    }
+
+    return false;
   }
 
   @Override
   public InputRow nextRow()
   {
-    while (true) {
+    if (savedInputRow != null) {
+      final InputRow row = savedInputRow;
+      savedInputRow = null;
+      return row;
+    } else {
+      while (firehose.hasMore()) {
         final InputRow row = firehose.nextRow();
-      if (row.getTimestampFromEpoch() >= minTime.getMillis()) {
+        if (acceptable(row)) {
           return row;
         }
       }
+
+      throw new NoSuchElementException("No more rows!");
+    }
   }
 
   @Override
@@ -44,4 +71,9 @@ public class MinTimeFirehose implements Firehose
   {
     firehose.close();
   }
+
+  private boolean acceptable(InputRow row)
+  {
+    return row.getTimestampFromEpoch() >= minTime.getMillis();
+  }
 }