Merge pull request #692 from metamx/handle-arithmetic-ex

Handle arithmetic exception for overlow

server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java

@@ -72,10 +72,8 @@ import java.util.concurrent.ScheduledExecutorService;
 public class RealtimePlumber implements Plumber
 {
   private static final EmittingLogger log = new EmittingLogger(RealtimePlumber.class);
-
   private final DataSchema schema;
   private final RealtimeTuningConfig config;
-
   private final RejectionPolicy rejectionPolicy;
   private final FireDepartmentMetrics metrics;
   private final ServiceEmitter emitter;
@@ -625,10 +623,7 @@ public class RealtimePlumber implements Plumber
 
     final long windowMillis = windowPeriod.toStandardDuration().getMillis();
     log.info("Starting merge and push.");
-
-    DateTime minTimestampAsDate = segmentGranularity.truncate(
-        rejectionPolicy.getCurrMaxTime().minus(windowMillis)
-    );
+    DateTime minTimestampAsDate = new DateTime(Math.max(windowMillis, rejectionPolicy.getCurrMaxTime().getMillis()) - windowMillis);
     long minTimestamp = minTimestampAsDate.getMillis();
 
     log.info("Found [%,d] sinks. minTimestamp [%s]", sinks.size(), minTimestampAsDate);