diff --git a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java index 2ae1db05925..3469536df9e 100644 --- a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java +++ b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java @@ -608,7 +608,16 @@ public class RealtimePlumber implements Plumber final long windowMillis = windowPeriod.toStandardDuration().getMillis(); log.info("Starting merge and push."); - DateTime minTimestampAsDate = new DateTime(Math.max(windowMillis, rejectionPolicy.getCurrMaxTime().getMillis()) - windowMillis); + DateTime minTimestampAsDate = segmentGranularity.truncate( + new DateTime( + Math.max( + windowMillis, + rejectionPolicy.getCurrMaxTime() + .getMillis() + ) + - windowMillis + ) + ); long minTimestamp = minTimestampAsDate.getMillis(); log.info("Found [%,d] sinks. minTimestamp [%s]", sinks.size(), minTimestampAsDate);