From 948b19a08856697bf0360cdc33d5cbd869e3a8e3 Mon Sep 17 00:00:00 2001
From: Parag Jain <paragjain16@gmail.com>
Date: Tue, 15 Mar 2016 11:47:32 -0500
Subject: [PATCH] do not silently ingnore rows

---
 .../segment/realtime/firehose/PredicateFirehose.java   | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/server/src/main/java/io/druid/segment/realtime/firehose/PredicateFirehose.java b/server/src/main/java/io/druid/segment/realtime/firehose/PredicateFirehose.java
index c1fe824aea4..a3ca41d8f3e 100644
--- a/server/src/main/java/io/druid/segment/realtime/firehose/PredicateFirehose.java
+++ b/server/src/main/java/io/druid/segment/realtime/firehose/PredicateFirehose.java
@@ -20,6 +20,7 @@
 package io.druid.segment.realtime.firehose;
 
 import com.google.common.base.Predicate;
+import com.metamx.common.logger.Logger;
 import io.druid.data.input.Firehose;
 import io.druid.data.input.InputRow;
 
@@ -31,6 +32,10 @@ import java.io.IOException;
  */
 public class PredicateFirehose implements Firehose
 {
+  private static final Logger log = new Logger(PredicateFirehose.class);
+  private static final int IGNORE_THRESHOLD = 5000;
+  private long ignored = 0;
+
   private final Firehose firehose;
   private final Predicate<InputRow> predicate;
 
@@ -55,6 +60,11 @@ public class PredicateFirehose implements Firehose
         savedInputRow = row;
         return true;
       }
+      // Do not silently discard the rows
+      if (ignored % IGNORE_THRESHOLD == 0) {
+        log.warn("[%,d] InputRow(s) ignored as they do not satisfy the predicate", ignored);
+      }
+      ignored++;
     }
 
     return false;