From 9a3ed353dc734fd80a107ac2051e7d7188f48e0b Mon Sep 17 00:00:00 2001
From: fjy <fangjinyang@gmail.com>
Date: Wed, 25 Feb 2015 16:05:51 -0800
Subject: [PATCH] Have a more meaningful message when events are dropped or
 unparseable

---
 .../realtime/RealtimeMetricsMonitor.java        | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/server/src/main/java/io/druid/segment/realtime/RealtimeMetricsMonitor.java b/server/src/main/java/io/druid/segment/realtime/RealtimeMetricsMonitor.java
index 57c0e3444df..2b4afca34d0 100644
--- a/server/src/main/java/io/druid/segment/realtime/RealtimeMetricsMonitor.java
+++ b/server/src/main/java/io/druid/segment/realtime/RealtimeMetricsMonitor.java
@@ -19,6 +19,7 @@ package io.druid.segment.realtime;
 
 import com.google.common.collect.Maps;
 import com.google.inject.Inject;
+import com.metamx.emitter.EmittingLogger;
 import com.metamx.emitter.service.ServiceEmitter;
 import com.metamx.emitter.service.ServiceMetricEvent;
 import com.metamx.metrics.AbstractMonitor;
@@ -30,6 +31,8 @@ import java.util.Map;
  */
 public class RealtimeMetricsMonitor extends AbstractMonitor
 {
+  private static final EmittingLogger log = new EmittingLogger(RealtimeMetricsMonitor.class);
+
   private final Map<FireDepartment, FireDepartmentMetrics> previousValues;
   private final List<FireDepartment> fireDepartments;
 
@@ -37,7 +40,7 @@ public class RealtimeMetricsMonitor extends AbstractMonitor
   public RealtimeMetricsMonitor(List<FireDepartment> fireDepartments)
   {
     this.fireDepartments = fireDepartments;
-    previousValues = Maps.newHashMap();
+    this.previousValues = Maps.newHashMap();
   }
 
   @Override
@@ -54,8 +57,16 @@ public class RealtimeMetricsMonitor extends AbstractMonitor
       final ServiceMetricEvent.Builder builder = new ServiceMetricEvent.Builder()
           .setUser2(fireDepartment.getDataSchema().getDataSource());
 
-      emitter.emit(builder.build("events/thrownAway", metrics.thrownAway() - previous.thrownAway()));
-      emitter.emit(builder.build("events/unparseable", metrics.unparseable() - previous.unparseable()));
+      final long thrownAway = metrics.thrownAway() - previous.thrownAway();
+      if (thrownAway > 0) {
+        log.warn("[%,d] events thrown away because they are outside the window period!", thrownAway);
+      }
+      emitter.emit(builder.build("events/thrownAway", thrownAway));
+      final long unparseable = metrics.unparseable() - previous.unparseable();
+      if (unparseable > 0) {
+        log.error("[%,d] Unparseable events! Turn on debug logging to see exception stack trace.", unparseable);
+      }
+      emitter.emit(builder.build("events/unparseable", unparseable));
       emitter.emit(builder.build("events/processed", metrics.processed() - previous.processed()));
       emitter.emit(builder.build("rows/output", metrics.rowOutput() - previous.rowOutput()));
       emitter.emit(builder.build("persists/num", metrics.numPersists() - previous.numPersists()));