diff --git a/server/src/main/java/org/elasticsearch/common/rounding/javatime/Rounding.java b/server/src/main/java/org/elasticsearch/common/rounding/javatime/Rounding.java
new file mode 100644
index 0000000000000..a36194dfdf45a
--- /dev/null
+++ b/server/src/main/java/org/elasticsearch/common/rounding/javatime/Rounding.java
@@ -0,0 +1,595 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.common.rounding.javatime;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.unit.TimeValue;
+import org.joda.time.DateTimeZone;
+import org.joda.time.IllegalInstantException;
+
+import java.io.IOException;
+import java.time.DayOfWeek;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoField;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.IsoFields;
+import java.time.temporal.TemporalField;
+import java.time.zone.ZoneOffsetTransition;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A strategy for rounding long values.
+ */
+public abstract class Rounding implements Writeable {
+
+    public static String format(long epochMillis) {
+        return Instant.ofEpochMilli(epochMillis) + "/" + epochMillis;
+    }
+
+    public enum DateTimeUnit {
+        WEEK_OF_WEEKYEAR(   (byte) 1, IsoFields.WEEK_OF_WEEK_BASED_YEAR),
+        YEAR_OF_CENTURY(    (byte) 2, ChronoField.YEAR_OF_ERA),
+        QUARTER_OF_YEAR(    (byte) 3, IsoFields.QUARTER_OF_YEAR),
+        MONTH_OF_YEAR(      (byte) 4, ChronoField.MONTH_OF_YEAR),
+        DAY_OF_MONTH(       (byte) 5, ChronoField.DAY_OF_MONTH),
+        HOUR_OF_DAY(        (byte) 6, ChronoField.HOUR_OF_DAY),
+        MINUTES_OF_HOUR(    (byte) 7, ChronoField.MINUTE_OF_HOUR),
+        SECOND_OF_MINUTE(   (byte) 8, ChronoField.SECOND_OF_MINUTE);
+
+        private final byte id;
+        private final TemporalField field;
+
+        DateTimeUnit(byte id, TemporalField field) {
+            this.id = id;
+            this.field = field;
+        }
+
+        public byte getId() {
+            return id;
+        }
+
+        public TemporalField getField() {
+            return field;
+        }
+
+        public static DateTimeUnit resolve(byte id) {
+            switch (id) {
+                case 1: return WEEK_OF_WEEKYEAR;
+                case 2: return YEAR_OF_CENTURY;
+                case 3: return QUARTER_OF_YEAR;
+                case 4: return MONTH_OF_YEAR;
+                case 5: return DAY_OF_MONTH;
+                case 6: return HOUR_OF_DAY;
+                case 7: return MINUTES_OF_HOUR;
+                case 8: return SECOND_OF_MINUTE;
+                default: throw new ElasticsearchException("Unknown date time unit id [" + id + "]");
+            }
+        }
+
+    }
+
+    public abstract byte id();
+
+    /**
+     * Rounds the given value.
+     */
+    public abstract long round(long value);
+
+    /**
+     * Given the rounded value (which was potentially generated by {@link #round(long)}, returns the next rounding value. For example, with
+     * interval based rounding, if the interval is 3, {@code nextRoundValue(6) = 9 }.
+     *
+     * @param value The current rounding value
+     * @return The next rounding value
+     */
+    public abstract long nextRoundingValue(long value);
+
+    @Override
+    public abstract boolean equals(Object obj);
+
+    @Override
+    public abstract int hashCode();
+
+    public static Builder builder(DateTimeUnit unit) {
+        return new Builder(unit);
+    }
+
+    public static Builder builder(TimeValue interval) {
+        return new Builder(interval);
+    }
+
+    public static class Builder {
+
+        private final DateTimeUnit unit;
+        private final long interval;
+
+        private ZoneId timeZone = ZoneOffset.UTC;
+
+        public Builder(DateTimeUnit unit) {
+            this.unit = unit;
+            this.interval = -1;
+        }
+
+        public Builder(TimeValue interval) {
+            this.unit = null;
+            if (interval.millis() < 1)
+                throw new IllegalArgumentException("Zero or negative time interval not supported");
+            this.interval = interval.millis();
+        }
+
+        public Builder timeZone(ZoneId timeZone) {
+            if (timeZone == null) {
+                throw new IllegalArgumentException("Setting null as timezone is not supported");
+            }
+            this.timeZone = timeZone;
+            return this;
+        }
+
+        public Rounding build() {
+            Rounding timeZoneRounding;
+            if (unit != null) {
+                timeZoneRounding = new TimeUnitRounding(unit, timeZone);
+            } else {
+                timeZoneRounding = new TimeIntervalRounding(interval, timeZone);
+            }
+            return timeZoneRounding;
+        }
+    }
+
+    static class TimeUnitRounding extends Rounding {
+
+        static final byte ID = 1;
+
+        private final DateTimeUnit unit;
+        private final ZoneId timeZone;
+        private final boolean unitRoundsToMidnight;
+
+
+        TimeUnitRounding(DateTimeUnit unit, ZoneId timeZone) {
+            this.unit = unit;
+            this.unitRoundsToMidnight = this.unit.field.getBaseUnit().getDuration().toMillis() > 60L * 60L * 1000L;
+            this.timeZone = timeZone;
+        }
+
+        TimeUnitRounding(StreamInput in) throws IOException {
+            unit = DateTimeUnit.resolve(in.readByte());
+            timeZone = ZoneId.of(in.readString());
+            unitRoundsToMidnight = unit.getField().getBaseUnit().getDuration().toMillis() > 60L * 60L * 1000L;
+        }
+
+        @Override
+        public byte id() {
+            return ID;
+        }
+
+        private LocalDateTime truncateLocalDateTime(LocalDateTime localDateTime) {
+            localDateTime = localDateTime.withNano(0);
+            assert localDateTime.getNano() == 0;
+            if (unit.equals(DateTimeUnit.SECOND_OF_MINUTE)) {
+                return localDateTime;
+            }
+
+            localDateTime = localDateTime.withSecond(0);
+            assert localDateTime.getSecond() == 0;
+            if (unit.equals(DateTimeUnit.MINUTES_OF_HOUR)) {
+                return localDateTime;
+            }
+
+            localDateTime = localDateTime.withMinute(0);
+            assert localDateTime.getMinute() == 0;
+            if (unit.equals(DateTimeUnit.HOUR_OF_DAY)) {
+                return localDateTime;
+            }
+
+            localDateTime = localDateTime.withHour(0);
+            assert localDateTime.getHour() == 0;
+            if (unit.equals(DateTimeUnit.DAY_OF_MONTH)) {
+                return localDateTime;
+            }
+
+            if (unit.equals(DateTimeUnit.WEEK_OF_WEEKYEAR)) {
+                localDateTime = localDateTime.with(ChronoField.DAY_OF_WEEK, 1);
+                assert localDateTime.getDayOfWeek() == DayOfWeek.MONDAY;
+                return localDateTime;
+            }
+
+            localDateTime = localDateTime.withDayOfMonth(1);
+            assert localDateTime.getDayOfMonth() == 1;
+            if (unit.equals(DateTimeUnit.MONTH_OF_YEAR)) {
+                return localDateTime;
+            }
+
+            if (unit.equals(DateTimeUnit.QUARTER_OF_YEAR)) {
+                int quarter = (int) IsoFields.QUARTER_OF_YEAR.getFrom(localDateTime);
+                int month = ((quarter - 1) * 3) + 1;
+                localDateTime = localDateTime.withMonth(month);
+                assert localDateTime.getMonthValue() % 3 == 1;
+                return localDateTime;
+            }
+
+            if (unit.equals(DateTimeUnit.YEAR_OF_CENTURY)) {
+                localDateTime = localDateTime.withMonth(1);
+                assert localDateTime.getMonthValue() == 1;
+                return localDateTime;
+            }
+
+            throw new IllegalArgumentException("NOT YET IMPLEMENTED for unit " + unit);
+        }
+
+        @Override
+        public long round(long utcMillis) {
+            if (unitRoundsToMidnight) {
+                final ZonedDateTime zonedDateTime = Instant.ofEpochMilli(utcMillis).atZone(timeZone);
+                final LocalDateTime localDateTime = zonedDateTime.toLocalDateTime();
+                final LocalDateTime localMidnight = truncateLocalDateTime(localDateTime);
+                return firstTimeOnDay(localMidnight);
+            } else {
+                while (true) {
+                    final Instant truncatedTime = truncateAsLocalTime(utcMillis);
+                    final ZoneOffsetTransition previousTransition = timeZone.getRules().previousTransition(Instant.ofEpochMilli(utcMillis));
+
+                    if (previousTransition == null) {
+                        // truncateAsLocalTime cannot have failed if there were no previous transitions
+                        return truncatedTime.toEpochMilli();
+                    }
+
+                    final long previousTransitionMillis = previousTransition.getInstant().toEpochMilli();
+
+                    if (truncatedTime != null && previousTransitionMillis <= truncatedTime.toEpochMilli()) {
+                        return truncatedTime.toEpochMilli();
+                    }
+
+                    // There was a transition in between the input time and the truncated time. Return to the transition time and
+                    // round that down instead.
+                    utcMillis = previousTransitionMillis - 1;
+                }
+            }
+        }
+
+        private long firstTimeOnDay(LocalDateTime localMidnight) {
+            assert localMidnight.toLocalTime().equals(LocalTime.of(0, 0, 0)) : "firstTimeOnDay should only be called at midnight";
+            assert unitRoundsToMidnight : "firstTimeOnDay should only be called if unitRoundsToMidnight";
+
+            // Now work out what localMidnight actually means
+            final List<ZoneOffset> currentOffsets = timeZone.getRules().getValidOffsets(localMidnight);
+            if (currentOffsets.size() >= 1) {
+                // There is at least one midnight on this day, so choose the first
+                final ZoneOffset firstOffset = currentOffsets.get(0);
+                final OffsetDateTime offsetMidnight = localMidnight.atOffset(firstOffset);
+                return offsetMidnight.toInstant().toEpochMilli();
+            } else {
+                // There were no midnights on this day, so we must have entered the day via an offset transition.
+                // Use the time of the transition as it is the earliest time on the right day.
+                ZoneOffsetTransition zoneOffsetTransition = timeZone.getRules().getTransition(localMidnight);
+                return zoneOffsetTransition.getInstant().toEpochMilli();
+            }
+        }
+
+        private Instant truncateAsLocalTime(long utcMillis) {
+            assert unitRoundsToMidnight == false : "truncateAsLocalTime should not be called if unitRoundsToMidnight";
+
+            final LocalDateTime truncatedLocalDateTime
+                = truncateLocalDateTime(Instant.ofEpochMilli(utcMillis).atZone(timeZone).toLocalDateTime());
+            final List<ZoneOffset> currentOffsets = timeZone.getRules().getValidOffsets(truncatedLocalDateTime);
+
+            if (currentOffsets.size() >= 1) {
+                // at least one possibilities - choose the latest one that's still no later than the input time
+                for (int offsetIndex = currentOffsets.size() - 1; offsetIndex >= 0; offsetIndex--) {
+                    final Instant result = truncatedLocalDateTime.atOffset(currentOffsets.get(offsetIndex)).toInstant();
+                    if (result.toEpochMilli() <= utcMillis) {
+                        return result;
+                    }
+                }
+
+                assert false : "rounded time not found for " + utcMillis + " with " + this;
+                return null;
+            } else {
+                // The chosen local time didn't happen. This means we were given a time in an hour (or a minute) whose start
+                // is missing due to an offset transition, so the time cannot be truncated.
+                return null;
+            }
+        }
+
+        private LocalDateTime nextRelevantMidnight(LocalDateTime localMidnight) {
+            assert localMidnight.toLocalTime().equals(LocalTime.of(0, 0, 0)) : "nextRelevantMidnight should only be called at midnight";
+            assert unitRoundsToMidnight : "firstTimeOnDay should only be called if unitRoundsToMidnight";
+
+            switch (unit) {
+                case DAY_OF_MONTH:
+                    return localMidnight.plus(1, ChronoUnit.DAYS);
+                case WEEK_OF_WEEKYEAR:
+                    return localMidnight.plus(7, ChronoUnit.DAYS);
+                case MONTH_OF_YEAR:
+                    return localMidnight.plus(1, ChronoUnit.MONTHS);
+                case QUARTER_OF_YEAR:
+                    return localMidnight.plus(3, ChronoUnit.MONTHS);
+                case YEAR_OF_CENTURY:
+                    return localMidnight.plus(1, ChronoUnit.YEARS);
+                default:
+                    throw new IllegalArgumentException("Unknown round-to-midnight unit: " + unit);
+            }
+        }
+
+        @Override
+        public long nextRoundingValue(long utcMillis) {
+            if (unitRoundsToMidnight) {
+                final ZonedDateTime zonedDateTime = Instant.ofEpochMilli(utcMillis).atZone(timeZone);
+                final LocalDateTime localDateTime = zonedDateTime.toLocalDateTime();
+                final LocalDateTime earlierLocalMidnight = truncateLocalDateTime(localDateTime);
+                final LocalDateTime localMidnight = nextRelevantMidnight(earlierLocalMidnight);
+                return firstTimeOnDay(localMidnight);
+            } else {
+                final long unitSize = unit.field.getBaseUnit().getDuration().toMillis();
+                final long roundedAfterOneIncrement = round(utcMillis + unitSize);
+                if (utcMillis < roundedAfterOneIncrement) {
+                    return roundedAfterOneIncrement;
+                } else {
+                    return round(utcMillis + 2 * unitSize);
+                }
+            }
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeByte(unit.getId());
+            String tz = ZoneOffset.UTC.equals(timeZone) ? "UTC" : timeZone.getId(); // stay joda compatible
+            out.writeString(tz);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(unit, timeZone);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            TimeUnitRounding other = (TimeUnitRounding) obj;
+            return Objects.equals(unit, other.unit) && Objects.equals(timeZone, other.timeZone);
+        }
+
+        @Override
+        public String toString() {
+            return "[" + timeZone + "][" + unit + "]";
+        }
+    }
+
+    static class TimeIntervalRounding extends Rounding {
+
+        static final byte ID = 2;
+
+        private final long interval;
+        private final ZoneId timeZone;
+
+        TimeIntervalRounding(long interval, ZoneId timeZone) {
+            if (interval < 1)
+                throw new IllegalArgumentException("Zero or negative time interval not supported");
+            this.interval = interval;
+            this.timeZone = timeZone;
+        }
+
+        TimeIntervalRounding(StreamInput in) throws IOException {
+            interval = in.readVLong();
+            timeZone = ZoneId.of(in.readString());
+        }
+
+        @Override
+        public byte id() {
+            return ID;
+        }
+
+        @Override
+        public long round(final long utcMillis) {
+            final Instant utcInstant = Instant.ofEpochMilli(utcMillis);
+            int offsetSeconds = timeZone.getRules().getOffset(utcInstant).getTotalSeconds();
+            long timeLocal = utcMillis + (offsetSeconds * 1000);
+
+            final long rounded = roundKey(timeLocal, interval) * interval;
+            long roundedUTC;
+            if (isInDSTGap(rounded) == false) {
+                int offsetOriginal = timeZone.getRules().getOffset(utcInstant).getTotalSeconds();
+                long instantUtc = rounded - (offsetOriginal * 1000);
+                int offsetLocalFromOriginal = timeZone.getRules().getOffset(Instant.ofEpochMilli(instantUtc)).getTotalSeconds();
+                if (offsetLocalFromOriginal == offsetOriginal) {
+                    roundedUTC = instantUtc;
+                } else {
+                    int offsetLocal = timeZone.getRules().getOffset(Instant.ofEpochMilli(rounded)).getTotalSeconds();
+                    int offset = timeZone.getRules().getOffset(Instant.ofEpochMilli(rounded - (offsetLocal * 1000))).getTotalSeconds();
+                    if (offsetLocal != offset) {
+                        long nextLocal = timeZone.getRules()
+                            .nextTransition(Instant.ofEpochMilli(rounded - (offsetLocal * 1000))).getInstant().toEpochMilli();
+                        if (nextLocal == (rounded - (offsetLocal * 1000))) {
+                            nextLocal = Long.MAX_VALUE;
+                        }
+                        long nextAdjusted = timeZone.getRules()
+                            .nextTransition(Instant.ofEpochMilli(rounded - (offset * 1000))).getInstant().toEpochMilli();
+                        if (nextAdjusted == (rounded - (offset * 1000))) {
+                            nextAdjusted = Long.MAX_VALUE;
+                        }
+                        if (nextAdjusted != nextLocal) {
+                            offset = offsetLocal;
+                        }
+                    }
+                    roundedUTC = rounded - (offset * 1000);
+                }
+                // check if we crossed DST transition, in this case we want the
+                // last rounded value before the transition
+
+                ZoneOffsetTransition offsetTransition = timeZone.getRules().previousTransition(Instant.ofEpochMilli(utcMillis+1));
+                if (offsetTransition != null) {
+                    long transition = offsetTransition.getInstant().toEpochMilli() - 1;
+                    if (transition != utcMillis && transition > roundedUTC) {
+                            roundedUTC = round(transition);
+                    }
+                }
+            } else {
+                /*
+                 * Edge case where the rounded local time is illegal and landed
+                 * in a DST gap. In this case, we choose 1ms tick after the
+                 * transition date. We don't want the transition date itself
+                 * because those dates, when rounded themselves, fall into the
+                 * previous interval. This would violate the invariant that the
+                 * rounding operation should be idempotent.
+                 */
+                ZoneOffsetTransition transition = timeZone.getRules().previousTransition(utcInstant);
+                if (transition != null) {
+                    long previousTransition = transition.getInstant().toEpochMilli();
+                    // if the previous transition is older than what we rounded to, we cannot use it as we would go too far in the past
+                    if (previousTransition < rounded) {
+                        roundedUTC = utcMillis;
+                    } else {
+                        roundedUTC = previousTransition + 1;
+                    }
+                } else {
+                    roundedUTC = utcMillis;
+                }
+            }
+            return roundedUTC;
+        }
+
+        private static long roundKey(long value, long interval) {
+            if (value < 0) {
+                return (value - interval + 1) / interval;
+            } else {
+                return value / interval;
+            }
+        }
+
+        /**
+         * Determine whether the local instant is a valid instant in the given
+         * time zone. The logic for this is taken from
+         * {@link DateTimeZone#convertLocalToUTC(long, boolean)} for the
+         * `strict` mode case, but instead of throwing an
+         * {@link IllegalInstantException}, which is costly, we want to return a
+         * flag indicating that the value is illegal in that time zone.
+         */
+        private boolean isInDSTGap(long instantLocal) {
+            if (timeZone.getRules().isFixedOffset()) {
+                return false;
+            }
+
+            // get the offset at instantLocal (first estimate)
+            int offsetLocal = timeZone.getRules().getOffset(Instant.ofEpochMilli(instantLocal)).getTotalSeconds() * 1000;
+            // adjust instantLocal using the estimate and recalc the offset
+            int offset = timeZone.getRules().getOffset(Instant.ofEpochMilli(instantLocal - offsetLocal)).getTotalSeconds() * 1000;
+            // if the offsets differ, we must be near a DST boundary
+            if (offsetLocal != offset) {
+                // determine if we are in the DST gap
+                long nextLocal = Long.MIN_VALUE;
+                ZoneOffsetTransition nextLocalTransition = timeZone.getRules()
+                    .nextTransition(Instant.ofEpochMilli(instantLocal - offsetLocal + 1));
+                if (nextLocalTransition != null) {
+                    nextLocal = nextLocalTransition.getInstant().toEpochMilli();
+                    if (nextLocal == (instantLocal - offsetLocal)) {
+                        nextLocal = Long.MAX_VALUE;
+                    }
+                }
+                long nextAdjusted = Long.MIN_VALUE;
+                ZoneOffsetTransition zoneOffsetTransition = timeZone.getRules()
+                    .nextTransition(Instant.ofEpochMilli(instantLocal - offset + 1));
+                if (zoneOffsetTransition != null) {
+                    nextAdjusted = zoneOffsetTransition.getInstant().toEpochMilli();
+                    if (zoneOffsetTransition == null || nextAdjusted == (instantLocal - offset)) {
+                        nextAdjusted = Long.MAX_VALUE;
+                    }
+                }
+                if (nextLocal != nextAdjusted) {
+                    // we are in the DST gap
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        @Override
+        public long nextRoundingValue(long time) {
+            int offsetSeconds = timeZone.getRules().getOffset(Instant.ofEpochMilli(time)).getTotalSeconds();
+            return ZonedDateTime.ofInstant(Instant.ofEpochMilli(time), ZoneOffset.UTC)
+                .plusSeconds(offsetSeconds)
+                .plusNanos(interval * 1_000_000)
+                .withZoneSameLocal(timeZone)
+                .toInstant().toEpochMilli();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeVLong(interval);
+            String tz = ZoneOffset.UTC.equals(timeZone) ? "UTC" : timeZone.getId(); // stay joda compatible
+            out.writeString(tz);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(interval, timeZone);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            TimeIntervalRounding other = (TimeIntervalRounding) obj;
+            return Objects.equals(interval, other.interval) && Objects.equals(timeZone, other.timeZone);
+        }
+    }
+
+    public static class Streams {
+
+        public static void write(Rounding rounding, StreamOutput out) throws IOException {
+            out.writeByte(rounding.id());
+            rounding.writeTo(out);
+        }
+
+        public static Rounding read(StreamInput in) throws IOException {
+            Rounding rounding;
+            byte id = in.readByte();
+            switch (id) {
+                case TimeUnitRounding.ID:
+                    rounding = new TimeUnitRounding(in);
+                    break;
+                case TimeIntervalRounding.ID:
+                    rounding = new TimeIntervalRounding(in);
+                    break;
+                default:
+                    throw new ElasticsearchException("unknown rounding id [" + id + "]");
+            }
+            return rounding;
+        }
+
+    }
+
+}
diff --git a/server/src/test/java/org/elasticsearch/common/rounding/DateTimeUnitTests.java b/server/src/test/java/org/elasticsearch/common/rounding/DateTimeUnitTests.java
index 6c92d3d1f9a19..33ea83c759297 100644
--- a/server/src/test/java/org/elasticsearch/common/rounding/DateTimeUnitTests.java
+++ b/server/src/test/java/org/elasticsearch/common/rounding/DateTimeUnitTests.java
@@ -19,6 +19,11 @@
 package org.elasticsearch.common.rounding;
 
 import org.elasticsearch.test.ESTestCase;
+import org.joda.time.DateTimeZone;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
 
 import static org.elasticsearch.common.rounding.DateTimeUnit.DAY_OF_MONTH;
 import static org.elasticsearch.common.rounding.DateTimeUnit.HOUR_OF_DAY;
@@ -28,6 +33,7 @@
 import static org.elasticsearch.common.rounding.DateTimeUnit.SECOND_OF_MINUTE;
 import static org.elasticsearch.common.rounding.DateTimeUnit.WEEK_OF_WEEKYEAR;
 import static org.elasticsearch.common.rounding.DateTimeUnit.YEAR_OF_CENTURY;
+import static org.hamcrest.Matchers.is;
 
 public class DateTimeUnitTests extends ESTestCase {
 
@@ -59,4 +65,17 @@ public void testEnumIds() {
         assertEquals(8, SECOND_OF_MINUTE.id());
         assertEquals(SECOND_OF_MINUTE, DateTimeUnit.resolve((byte) 8));
     }
+
+    public void testConversion() {
+        long millis = randomLongBetween(0, Instant.now().toEpochMilli());
+        DateTimeZone zone = randomDateTimeZone();
+        ZoneId zoneId = ZoneId.of(zone.getID());
+
+        int offsetSeconds = zoneId.getRules().getOffset(Instant.ofEpochMilli(millis)).getTotalSeconds();
+        long parsedMillisJavaTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), zoneId)
+            .minusSeconds(offsetSeconds).toInstant().toEpochMilli();
+
+        long parsedMillisJodaTime = zone.convertLocalToUTC(millis, true);
+        assertThat(parsedMillisJavaTime, is(parsedMillisJodaTime));
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/common/rounding/RoundingDuelTests.java b/server/src/test/java/org/elasticsearch/common/rounding/RoundingDuelTests.java
new file mode 100644
index 0000000000000..91aa327e65622
--- /dev/null
+++ b/server/src/test/java/org/elasticsearch/common/rounding/RoundingDuelTests.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.rounding;
+
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.test.ESTestCase;
+
+import java.time.ZoneOffset;
+
+import static org.hamcrest.Matchers.is;
+
+public class RoundingDuelTests extends ESTestCase  {
+
+    // dont include nano/micro seconds as rounding would become zero then and throw an exception
+    private static final String[] ALLOWED_TIME_SUFFIXES = new String[]{"d", "h", "ms", "s", "m"};
+
+    public void testSerialization() throws Exception {
+        org.elasticsearch.common.rounding.javatime.Rounding.DateTimeUnit randomDateTimeUnit = randomFrom(org.elasticsearch.common.rounding.javatime.Rounding.DateTimeUnit.values());
+        org.elasticsearch.common.rounding.javatime.Rounding rounding;
+        if (randomBoolean()) {
+            rounding = org.elasticsearch.common.rounding.javatime.Rounding.builder(randomDateTimeUnit).timeZone(ZoneOffset.UTC).build();
+        } else {
+            rounding = org.elasticsearch.common.rounding.javatime.Rounding.builder(timeValue()).timeZone(ZoneOffset.UTC).build();
+        }
+        BytesStreamOutput output = new BytesStreamOutput();
+        org.elasticsearch.common.rounding.javatime.Rounding.Streams.write(rounding, output);
+
+        org.elasticsearch.common.rounding.Rounding roundingJoda = org.elasticsearch.common.rounding.Rounding.Streams.read(output.bytes().streamInput());
+        org.elasticsearch.common.rounding.javatime.Rounding roundingJavaTime = org.elasticsearch.common.rounding.javatime.Rounding.Streams.read(output.bytes().streamInput());
+
+        int randomInt = randomIntBetween(1, 1_000_000_000);
+        assertThat(roundingJoda.round(randomInt), is(roundingJavaTime.round(randomInt)));
+        assertThat(roundingJoda.nextRoundingValue(randomInt), is(roundingJavaTime.nextRoundingValue(randomInt)));
+    }
+
+    static TimeValue timeValue() {
+        return TimeValue.parseTimeValue(randomIntBetween(1, 1000) + randomFrom(ALLOWED_TIME_SUFFIXES), "settingName");
+    }
+}
diff --git a/server/src/test/java/org/elasticsearch/common/rounding/javatime/RoundingTests.java b/server/src/test/java/org/elasticsearch/common/rounding/javatime/RoundingTests.java
new file mode 100644
index 0000000000000..0a6674f406fad
--- /dev/null
+++ b/server/src/test/java/org/elasticsearch/common/rounding/javatime/RoundingTests.java
@@ -0,0 +1,759 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.rounding.javatime;
+
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.rounding.DateTimeUnit;
+import org.elasticsearch.common.time.DateFormatters;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeMatcher;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.TemporalAccessor;
+import java.time.zone.ZoneOffsetTransition;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+public class RoundingTests extends ESTestCase {
+
+    public void testUTCTimeUnitRounding() {
+        Rounding tzRounding = Rounding.builder(Rounding.DateTimeUnit.MONTH_OF_YEAR).build();
+        ZoneId tz = ZoneOffset.UTC;
+        assertThat(tzRounding.round(time("2009-02-03T01:01:01")), isDate(time("2009-02-01T00:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-01T00:00:00.000Z")), isDate(time("2009-03-01T00:00:00.000Z"), tz));
+
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.WEEK_OF_WEEKYEAR).build();
+        assertThat(tzRounding.round(time("2012-01-10T01:01:01")), isDate(time("2012-01-09T00:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2012-01-09T00:00:00.000Z")), isDate(time("2012-01-16T00:00:00.000Z"), tz));
+    }
+
+    public void testUTCIntervalRounding() {
+        Rounding tzRounding = Rounding.builder(TimeValue.timeValueHours(12)).build();
+        ZoneId tz = ZoneOffset.UTC;
+        assertThat(tzRounding.round(time("2009-02-03T01:01:01")), isDate(time("2009-02-03T00:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T00:00:00.000Z")), isDate(time("2009-02-03T12:00:00.000Z"), tz));
+        assertThat(tzRounding.round(time("2009-02-03T13:01:01")), isDate(time("2009-02-03T12:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T12:00:00.000Z")), isDate(time("2009-02-04T00:00:00.000Z"), tz));
+
+        tzRounding = Rounding.builder(TimeValue.timeValueHours(48)).build();
+        assertThat(tzRounding.round(time("2009-02-03T01:01:01")), isDate(time("2009-02-03T00:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T00:00:00.000Z")), isDate(time("2009-02-05T00:00:00.000Z"), tz));
+        assertThat(tzRounding.round(time("2009-02-05T13:01:01")), isDate(time("2009-02-05T00:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-05T00:00:00.000Z")), isDate(time("2009-02-07T00:00:00.000Z"), tz));
+    }
+
+    /**
+     * test TimeIntervalRounding, (interval &lt; 12h) with time zone shift
+     */
+    public void testTimeIntervalRounding() {
+        ZoneId tz = ZoneOffset.ofHours(-1);
+        Rounding tzRounding = Rounding.builder(TimeValue.timeValueHours(6)).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2009-02-03T00:01:01")), isDate(time("2009-02-02T19:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-02T19:00:00.000Z")), isDate(time("2009-02-03T01:00:00.000Z"), tz));
+
+        assertThat(tzRounding.round(time("2009-02-03T13:01:01")), isDate(time("2009-02-03T13:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T13:00:00.000Z")), isDate(time("2009-02-03T19:00:00.000Z"), tz));
+    }
+
+    /**
+     * test DayIntervalRounding, (interval &gt;= 12h) with time zone shift
+     */
+    public void testDayIntervalRounding() {
+        ZoneId tz = ZoneOffset.ofHours(-8);
+        Rounding tzRounding = Rounding.builder(TimeValue.timeValueHours(12)).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2009-02-03T00:01:01")), isDate(time("2009-02-02T20:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-02T20:00:00.000Z")), isDate(time("2009-02-03T08:00:00.000Z"), tz));
+
+        assertThat(tzRounding.round(time("2009-02-03T13:01:01")), isDate(time("2009-02-03T08:00:00.000Z"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T08:00:00.000Z")), isDate(time("2009-02-03T20:00:00.000Z"), tz));
+    }
+
+    public void testDayRounding() {
+        int timezoneOffset = -2;
+        Rounding tzRounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH)
+            .timeZone(ZoneOffset.ofHours(timezoneOffset)).build();
+        assertThat(tzRounding.round(0), equalTo(0L - TimeValue.timeValueHours(24 + timezoneOffset).millis()));
+        assertThat(tzRounding.nextRoundingValue(0L - TimeValue.timeValueHours(24 + timezoneOffset).millis()), equalTo(0L - TimeValue
+            .timeValueHours(timezoneOffset).millis()));
+
+        ZoneId tz = ZoneId.of("-08:00");
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2012-04-01T04:15:30Z")), isDate(time("2012-03-31T08:00:00Z"), tz));
+
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.MONTH_OF_YEAR).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2012-04-01T04:15:30Z")), equalTo(time("2012-03-01T08:00:00Z")));
+
+        // date in Feb-3rd, but still in Feb-2nd in -02:00 timezone
+        tz = ZoneId.of("-02:00");
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2009-02-03T01:01:01")), isDate(time("2009-02-02T02:00:00"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-02T02:00:00")), isDate(time("2009-02-03T02:00:00"), tz));
+
+        // date in Feb-3rd, also in -02:00 timezone
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2009-02-03T02:01:01")), isDate(time("2009-02-03T02:00:00"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T02:00:00")), isDate(time("2009-02-04T02:00:00"), tz));
+    }
+
+    public void testTimeRounding() {
+        // hour unit
+        ZoneId tz = ZoneOffset.ofHours(-2);
+        Rounding tzRounding = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY).timeZone(tz).build();
+        assertThat(tzRounding.round(0), equalTo(0L));
+        assertThat(tzRounding.nextRoundingValue(0L), equalTo(TimeValue.timeValueHours(1L).getMillis()));
+
+        assertThat(tzRounding.round(time("2009-02-03T01:01:01")), isDate(time("2009-02-03T01:00:00"), tz));
+        assertThat(tzRounding.nextRoundingValue(time("2009-02-03T01:00:00")), isDate(time("2009-02-03T02:00:00"), tz));
+    }
+
+    public void testTimeUnitRoundingDST() {
+        Rounding tzRounding;
+        // testing savings to non savings switch
+        ZoneId cet = ZoneId.of("CET");
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY).timeZone(cet).build();
+        assertThat(tzRounding.round(time("2014-10-26T01:01:01", cet)), isDate(time("2014-10-26T01:00:00+02:00"), cet));
+        assertThat(tzRounding.nextRoundingValue(time("2014-10-26T01:00:00", cet)),isDate(time("2014-10-26T02:00:00+02:00"), cet));
+        assertThat(tzRounding.nextRoundingValue(time("2014-10-26T02:00:00", cet)), isDate(time("2014-10-26T02:00:00+01:00"), cet));
+
+        // testing non savings to savings switch
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY).timeZone(cet).build();
+        assertThat(tzRounding.round(time("2014-03-30T01:01:01", cet)), isDate(time("2014-03-30T01:00:00+01:00"), cet));
+        assertThat(tzRounding.nextRoundingValue(time("2014-03-30T01:00:00", cet)), isDate(time("2014-03-30T03:00:00", cet), cet));
+        assertThat(tzRounding.nextRoundingValue(time("2014-03-30T03:00:00", cet)), isDate(time("2014-03-30T04:00:00", cet), cet));
+
+        // testing non savings to savings switch (America/Chicago)
+        ZoneId chg = ZoneId.of("America/Chicago");
+        Rounding tzRounding_utc = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY)
+            .timeZone(ZoneOffset.UTC).build();
+        assertThat(tzRounding.round(time("2014-03-09T03:01:01", chg)), isDate(time("2014-03-09T03:00:00", chg), chg));
+
+        Rounding tzRounding_chg = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY).timeZone(chg).build();
+        assertThat(tzRounding_chg.round(time("2014-03-09T03:01:01", chg)), isDate(time("2014-03-09T03:00:00", chg), chg));
+
+        // testing savings to non savings switch 2013 (America/Chicago)
+        assertThat(tzRounding_utc.round(time("2013-11-03T06:01:01", chg)), isDate(time("2013-11-03T06:00:00", chg), chg));
+        assertThat(tzRounding_chg.round(time("2013-11-03T06:01:01", chg)), isDate(time("2013-11-03T06:00:00", chg), chg));
+
+        // testing savings to non savings switch 2014 (America/Chicago)
+        assertThat(tzRounding_utc.round(time("2014-11-02T06:01:01", chg)), isDate(time("2014-11-02T06:00:00", chg), chg));
+        assertThat(tzRounding_chg.round(time("2014-11-02T06:01:01", chg)), isDate(time("2014-11-02T06:00:00", chg), chg));
+    }
+
+    /**
+     * Randomized test on TimeUnitRounding. Test uses random
+     * {@link DateTimeUnit} and {@link ZoneId} and often (50% of the time)
+     * chooses test dates that are exactly on or close to offset changes (e.g.
+     * DST) in the chosen time zone.
+     *
+     * It rounds the test date down and up and performs various checks on the
+     * rounding unit interval that is defined by this. Assumptions tested are
+     * described in
+     * {@link #assertInterval(long, long, long, Rounding, ZoneId)}
+     */
+    public void testRoundingRandom() {
+        for (int i = 0; i < 1000; ++i) {
+            Rounding.DateTimeUnit unit = randomFrom(Rounding.DateTimeUnit.values());
+            ZoneId tz = randomZone();
+            Rounding rounding = new Rounding.TimeUnitRounding(unit, tz);
+            long date = Math.abs(randomLong() % (2 * (long) 10e11)); // 1970-01-01T00:00:00Z - 2033-05-18T05:33:20.000+02:00
+            long unitMillis = unit.getField().getBaseUnit().getDuration().toMillis();
+            // FIXME this was copy pasted from the other impl and not used. breaks the nasty date actually gets assigned
+            if (randomBoolean()) {
+                nastyDate(date, tz, unitMillis);
+            }
+            final long roundedDate = rounding.round(date);
+            final long nextRoundingValue = rounding.nextRoundingValue(roundedDate);
+
+            assertInterval(roundedDate, date, nextRoundingValue, rounding, tz);
+
+            // check correct unit interval width for units smaller than a day, they should be fixed size except for transitions
+            if (unitMillis <= 86400 * 1000) {
+                // if the interval defined didn't cross timezone offset transition, it should cover unitMillis width
+                int offsetRounded = tz.getRules().getOffset(Instant.ofEpochMilli(roundedDate - 1)).getTotalSeconds();
+                int offsetNextValue = tz.getRules().getOffset(Instant.ofEpochMilli(nextRoundingValue + 1)).getTotalSeconds();
+                if (offsetRounded == offsetNextValue) {
+                    assertThat("unit interval width not as expected for [" + unit + "], [" + tz + "] at "
+                        + Instant.ofEpochMilli(roundedDate), nextRoundingValue - roundedDate, equalTo(unitMillis));
+                }
+            }
+        }
+    }
+
+    /**
+     * To be even more nasty, go to a transition in the selected time zone.
+     * In one third of the cases stay there, otherwise go half a unit back or forth
+     */
+    private static long nastyDate(long initialDate, ZoneId timezone, long unitMillis) {
+        ZoneOffsetTransition transition = timezone.getRules().nextTransition(Instant.ofEpochMilli(initialDate));
+        long date = initialDate;
+        if (transition != null) {
+            date = transition.getInstant().toEpochMilli();
+        }
+        if (randomBoolean()) {
+            return date + (randomLong() % unitMillis);  // positive and negative offset possible
+        } else {
+            return date;
+        }
+    }
+
+    /**
+     * test DST end with interval rounding
+     * CET: 25 October 2015, 03:00:00 clocks were turned backward 1 hour to 25 October 2015, 02:00:00 local standard time
+     */
+    public void testTimeIntervalCET_DST_End() {
+        long interval = TimeUnit.MINUTES.toMillis(20);
+        ZoneId tz = ZoneId.of("CET");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+
+        assertThat(rounding.round(time("2015-10-25T01:55:00+02:00")), isDate(time("2015-10-25T01:40:00+02:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T02:15:00+02:00")), isDate(time("2015-10-25T02:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T02:35:00+02:00")), isDate(time("2015-10-25T02:20:00+02:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T02:55:00+02:00")), isDate(time("2015-10-25T02:40:00+02:00"), tz));
+        // after DST shift
+        assertThat(rounding.round(time("2015-10-25T02:15:00+01:00")), isDate(time("2015-10-25T02:00:00+01:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T02:35:00+01:00")), isDate(time("2015-10-25T02:20:00+01:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T02:55:00+01:00")), isDate(time("2015-10-25T02:40:00+01:00"), tz));
+        assertThat(rounding.round(time("2015-10-25T03:15:00+01:00")), isDate(time("2015-10-25T03:00:00+01:00"), tz));
+    }
+
+    /**
+     * test DST start with interval rounding
+     * CET: 27 March 2016, 02:00:00 clocks were turned forward 1 hour to 27 March 2016, 03:00:00 local daylight time
+     */
+    public void testTimeIntervalCET_DST_Start() {
+        long interval = TimeUnit.MINUTES.toMillis(20);
+        ZoneId tz = ZoneId.of("CET");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+        // test DST start
+        assertThat(rounding.round(time("2016-03-27T01:55:00+01:00")), isDate(time("2016-03-27T01:40:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T02:00:00+01:00")), isDate(time("2016-03-27T03:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T03:15:00+02:00")), isDate(time("2016-03-27T03:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T03:35:00+02:00")), isDate(time("2016-03-27T03:20:00+02:00"), tz));
+    }
+
+    /**
+     * test DST start with offset not fitting interval, e.g. Asia/Kathmandu
+     * adding 15min on 1986-01-01T00:00:00 the interval from
+     * 1986-01-01T00:15:00+05:45 to 1986-01-01T00:20:00+05:45 to only be 5min
+     * long
+     */
+    public void testTimeInterval_Kathmandu_DST_Start() {
+        long interval = TimeUnit.MINUTES.toMillis(20);
+        ZoneId tz = ZoneId.of("Asia/Kathmandu");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+        assertThat(rounding.round(time("1985-12-31T23:55:00+05:30")), isDate(time("1985-12-31T23:40:00+05:30"), tz));
+        // TODO all are working except this one
+//        assertThat(rounding.round(time("1986-01-01T00:16:00+05:45")), isDate(time("1986-01-01T00:15:00+05:45"), tz));
+        assertThat(time("1986-01-01T00:15:00+05:45") - time("1985-12-31T23:40:00+05:30"), equalTo(TimeUnit.MINUTES.toMillis(20)));
+        assertThat(rounding.round(time("1986-01-01T00:26:00+05:45")), isDate(time("1986-01-01T00:20:00+05:45"), tz));
+        assertThat(time("1986-01-01T00:20:00+05:45") - time("1986-01-01T00:15:00+05:45"), equalTo(TimeUnit.MINUTES.toMillis(5)));
+        assertThat(rounding.round(time("1986-01-01T00:46:00+05:45")), isDate(time("1986-01-01T00:40:00+05:45"), tz));
+        assertThat(time("1986-01-01T00:40:00+05:45") - time("1986-01-01T00:20:00+05:45"), equalTo(TimeUnit.MINUTES.toMillis(20)));
+    }
+
+    /**
+     * Special test for intervals that don't fit evenly into rounding interval.
+     * In this case, when interval crosses DST transition point, rounding in local
+     * time can land in a DST gap which results in wrong UTC rounding values.
+     */
+    public void testIntervalRounding_NotDivisibleInteval() {
+        long interval = TimeUnit.MINUTES.toMillis(14);
+        ZoneId tz = ZoneId.of("CET");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+
+        assertThat(rounding.round(time("2016-03-27T01:41:00+01:00")), isDate(time("2016-03-27T01:30:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T01:51:00+01:00")), isDate(time("2016-03-27T01:44:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T01:59:00+01:00")), isDate(time("2016-03-27T01:58:00+01:00"), tz));
+        // TODO all are working except this one
+//        assertThat(rounding.round(time("2016-03-27T03:05:00+02:00")), isDate(time("2016-03-27T03:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T03:12:00+02:00")), isDate(time("2016-03-27T03:08:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T03:25:00+02:00")), isDate(time("2016-03-27T03:22:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T03:39:00+02:00")), isDate(time("2016-03-27T03:36:00+02:00"), tz));
+    }
+
+    /**
+     * Test for half day rounding intervals scrossing DST.
+     */
+    public void testIntervalRounding_HalfDay_DST() {
+        long interval = TimeUnit.HOURS.toMillis(12);
+        ZoneId tz = ZoneId.of("CET");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+
+        assertThat(rounding.round(time("2016-03-26T01:00:00+01:00")), isDate(time("2016-03-26T00:00:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-26T13:00:00+01:00")), isDate(time("2016-03-26T12:00:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T01:00:00+01:00")), isDate(time("2016-03-27T00:00:00+01:00"), tz));
+        assertThat(rounding.round(time("2016-03-27T13:00:00+02:00")), isDate(time("2016-03-27T12:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-28T01:00:00+02:00")), isDate(time("2016-03-28T00:00:00+02:00"), tz));
+        assertThat(rounding.round(time("2016-03-28T13:00:00+02:00")), isDate(time("2016-03-28T12:00:00+02:00"), tz));
+    }
+
+    /**
+     * randomized test on {@link org.elasticsearch.common.rounding.Rounding.TimeIntervalRounding} with random interval and time zone offsets
+     */
+    public void testIntervalRoundingRandom() {
+        for (int i = 0; i < 1000; i++) {
+            TimeUnit unit = randomFrom(TimeUnit.MINUTES, TimeUnit.HOURS, TimeUnit.DAYS);
+            long interval = unit.toMillis(randomIntBetween(1, 365));
+            ZoneId tz = randomZone();
+            Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+            long mainDate = Math.abs(randomLong() % (2 * (long) 10e11)); // 1970-01-01T00:00:00Z - 2033-05-18T05:33:20.000+02:00
+            if (randomBoolean()) {
+                mainDate = nastyDate(mainDate, tz, interval);
+            }
+            // check two intervals around date
+            long previousRoundedValue = Long.MIN_VALUE;
+            for (long date = mainDate - 2 * interval; date < mainDate + 2 * interval; date += interval / 2) {
+                try {
+                    final long roundedDate = rounding.round(date);
+                    final long nextRoundingValue = rounding.nextRoundingValue(roundedDate);
+                    assertThat("Rounding should be idempotent", roundedDate, equalTo(rounding.round(roundedDate)));
+                    assertThat("Rounded value smaller or equal than unrounded", roundedDate, lessThanOrEqualTo(date));
+                    assertThat("Values smaller than rounded value should round further down", rounding.round(roundedDate - 1),
+                        lessThan(roundedDate));
+                    assertThat("Rounding should be >= previous rounding value", roundedDate, greaterThanOrEqualTo(previousRoundedValue));
+
+                    if (tz.getRules().isFixedOffset()) {
+                        assertThat("NextRounding value should be greater than date", nextRoundingValue, greaterThan(roundedDate));
+                        assertThat("NextRounding value should be interval from rounded value", nextRoundingValue - roundedDate,
+                            equalTo(interval));
+                        assertThat("NextRounding value should be a rounded date", nextRoundingValue,
+                            equalTo(rounding.round(nextRoundingValue)));
+                    }
+                    previousRoundedValue = roundedDate;
+                } catch (AssertionError e) {
+                    ZonedDateTime dateTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(date), tz);
+                    ZonedDateTime previousRoundedValueDate = ZonedDateTime.ofInstant(Instant.ofEpochMilli(previousRoundedValue), tz);
+                    logger.error("Rounding error at {}/{}, timezone {}, interval: {} previousRoundedValue {}/{}", dateTime, date,
+                        tz, interval, previousRoundedValueDate, previousRoundedValue);
+                    throw e;
+                }
+            }
+        }
+    }
+
+    /**
+     * Test that rounded values are always greater or equal to last rounded value if date is increasing.
+     * The example covers an interval around 2011-10-30T02:10:00+01:00, time zone CET, interval: 2700000ms
+     */
+    public void testIntervalRoundingMonotonic_CET() {
+        long interval = TimeUnit.MINUTES.toMillis(45);
+        ZoneId tz = ZoneId.of("CET");
+        Rounding rounding = new Rounding.TimeIntervalRounding(interval, tz);
+        List<Tuple<String, String>> expectedDates = new ArrayList<>();
+        // first date is the date to be rounded, second the expected result
+        expectedDates.add(new Tuple<>("2011-10-30T01:40:00.000+02:00", "2011-10-30T01:30:00.000+02:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:02:30.000+02:00", "2011-10-30T01:30:00.000+02:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:25:00.000+02:00", "2011-10-30T02:15:00.000+02:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:47:30.000+02:00", "2011-10-30T02:15:00.000+02:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:10:00.000+01:00", "2011-10-30T02:15:00.000+02:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:32:30.000+01:00", "2011-10-30T02:15:00.000+01:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T02:55:00.000+01:00", "2011-10-30T02:15:00.000+01:00"));
+        expectedDates.add(new Tuple<>("2011-10-30T03:17:30.000+01:00", "2011-10-30T03:00:00.000+01:00"));
+
+        long previousDate = Long.MIN_VALUE;
+        for (Tuple<String, String> dates : expectedDates) {
+            final long roundedDate = rounding.round(time(dates.v1()));
+            assertThat(roundedDate, isDate(time(dates.v2()), tz));
+            assertThat(roundedDate, greaterThanOrEqualTo(previousDate));
+            previousDate = roundedDate;
+        }
+        // here's what this means for interval widths
+        assertEquals(TimeUnit.MINUTES.toMillis(45), time("2011-10-30T02:15:00.000+02:00") - time("2011-10-30T01:30:00.000+02:00"));
+        assertEquals(TimeUnit.MINUTES.toMillis(60), time("2011-10-30T02:15:00.000+01:00") - time("2011-10-30T02:15:00.000+02:00"));
+        assertEquals(TimeUnit.MINUTES.toMillis(45), time("2011-10-30T03:00:00.000+01:00") - time("2011-10-30T02:15:00.000+01:00"));
+    }
+
+    /**
+     * special test for DST switch from #9491
+     */
+    public void testAmbiguousHoursAfterDSTSwitch() {
+        Rounding tzRounding;
+        final ZoneId tz = ZoneId.of("Asia/Jerusalem");
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.HOUR_OF_DAY).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2014-10-26T00:30:00+03:00")), isDate(time("2014-10-26T00:00:00+03:00"), tz));
+        assertThat(tzRounding.round(time("2014-10-26T01:30:00+03:00")), isDate(time("2014-10-26T01:00:00+03:00"), tz));
+        // the utc date for "2014-10-25T03:00:00+03:00" and "2014-10-25T03:00:00+02:00" is the same, local time turns back 1h here
+        assertThat(time("2014-10-26T03:00:00+03:00"), isDate(time("2014-10-26T02:00:00+02:00"), tz));
+        assertThat(tzRounding.round(time("2014-10-26T01:30:00+02:00")), isDate(time("2014-10-26T01:00:00+02:00"), tz));
+        assertThat(tzRounding.round(time("2014-10-26T02:30:00+02:00")), isDate(time("2014-10-26T02:00:00+02:00"), tz));
+
+        // Day interval
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2014-11-11T17:00:00", tz)), isDate(time("2014-11-11T00:00:00", tz), tz));
+        // DST on
+        assertThat(tzRounding.round(time("2014-08-11T17:00:00", tz)), isDate(time("2014-08-11T00:00:00", tz), tz));
+        // Day of switching DST on -> off
+        assertThat(tzRounding.round(time("2014-10-26T17:00:00", tz)), isDate(time("2014-10-26T00:00:00", tz), tz));
+        // Day of switching DST off -> on
+        assertThat(tzRounding.round(time("2015-03-27T17:00:00", tz)), isDate(time("2015-03-27T00:00:00", tz), tz));
+
+        // Month interval
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.MONTH_OF_YEAR).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2014-11-11T17:00:00", tz)), isDate(time("2014-11-01T00:00:00", tz), tz));
+        // DST on
+        assertThat(tzRounding.round(time("2014-10-10T17:00:00", tz)), isDate(time("2014-10-01T00:00:00", tz), tz));
+
+        // Year interval
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.YEAR_OF_CENTURY).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2014-11-11T17:00:00", tz)), isDate(time("2014-01-01T00:00:00", tz), tz));
+
+        // Two timestamps in same year and different timezone offset ("Double buckets" issue - #9491)
+        tzRounding = Rounding.builder(Rounding.DateTimeUnit.YEAR_OF_CENTURY).timeZone(tz).build();
+        assertThat(tzRounding.round(time("2014-11-11T17:00:00", tz)),
+            isDate(tzRounding.round(time("2014-08-11T17:00:00", tz)), tz));
+    }
+
+    /**
+     * test for #10025, strict local to UTC conversion can cause joda exceptions
+     * on DST start
+     */
+    public void testLenientConversionDST() {
+        ZoneId tz = ZoneId.of("America/Sao_Paulo");
+
+        long start = time("2014-10-18T20:50:00.000", tz);
+        long end = time("2014-10-19T01:00:00.000", tz);
+        Rounding tzRounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.MINUTES_OF_HOUR, tz);
+        Rounding dayTzRounding = new Rounding.TimeIntervalRounding(60000, tz);
+        for (long time = start; time < end; time = time + 60000) {
+            assertThat(tzRounding.nextRoundingValue(time), greaterThan(time));
+            assertThat(dayTzRounding.nextRoundingValue(time), greaterThan(time));
+        }
+    }
+
+    public void testEdgeCasesTransition() {
+        {
+            // standard +/-1 hour DST transition, CET
+            ZoneId tz = ZoneId.of("CET");
+            Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.HOUR_OF_DAY, tz);
+
+            // 29 Mar 2015 - Daylight Saving Time Started
+            // at 02:00:00 clocks were turned forward 1 hour to 03:00:00
+            assertInterval(time("2015-03-29T00:00:00.000+01:00"), time("2015-03-29T01:00:00.000+01:00"), rounding, 60, tz);
+            assertInterval(time("2015-03-29T01:00:00.000+01:00"), time("2015-03-29T03:00:00.000+02:00"), rounding, 60, tz);
+            assertInterval(time("2015-03-29T03:00:00.000+02:00"), time("2015-03-29T04:00:00.000+02:00"), rounding, 60, tz);
+
+            // 25 Oct 2015 - Daylight Saving Time Ended
+            // at 03:00:00 clocks were turned backward 1 hour to 02:00:00
+            assertInterval(time("2015-10-25T01:00:00.000+02:00"), time("2015-10-25T02:00:00.000+02:00"), rounding, 60, tz);
+            assertInterval(time("2015-10-25T02:00:00.000+02:00"), time("2015-10-25T02:00:00.000+01:00"), rounding, 60, tz);
+            assertInterval(time("2015-10-25T02:00:00.000+01:00"), time("2015-10-25T03:00:00.000+01:00"), rounding, 60, tz);
+        }
+
+        {
+            // time zone "Asia/Kathmandu"
+            // 1 Jan 1986 - Time Zone Change (IST → NPT), at 00:00:00 clocks were turned forward 00:15 minutes
+            //
+            // hour rounding is stable before 1985-12-31T23:00:00.000 and after 1986-01-01T01:00:00.000+05:45
+            // the interval between is 105 minutes long because the hour after transition starts at 00:15
+            // which is not a round value for hourly rounding
+            ZoneId tz = ZoneId.of("Asia/Kathmandu");
+            Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.HOUR_OF_DAY, tz);
+
+            assertInterval(time("1985-12-31T22:00:00.000+05:30"), time("1985-12-31T23:00:00.000+05:30"), rounding, 60, tz);
+            assertInterval(time("1985-12-31T23:00:00.000+05:30"), time("1986-01-01T01:00:00.000+05:45"), rounding, 105, tz);
+            assertInterval(time("1986-01-01T01:00:00.000+05:45"), time("1986-01-01T02:00:00.000+05:45"), rounding, 60, tz);
+        }
+
+        {
+            // time zone "Australia/Lord_Howe"
+            // 3 Mar 1991 - Daylight Saving Time Ended
+            // at 02:00:00 clocks were turned backward 0:30 hours to Sunday, 3 March 1991, 01:30:00
+            ZoneId tz = ZoneId.of("Australia/Lord_Howe");
+            Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.HOUR_OF_DAY, tz);
+
+            assertInterval(time("1991-03-03T00:00:00.000+11:00"), time("1991-03-03T01:00:00.000+11:00"), rounding, 60, tz);
+            assertInterval(time("1991-03-03T01:00:00.000+11:00"), time("1991-03-03T02:00:00.000+10:30"), rounding, 90, tz);
+            assertInterval(time("1991-03-03T02:00:00.000+10:30"), time("1991-03-03T03:00:00.000+10:30"), rounding, 60, tz);
+
+            // 27 Oct 1991 - Daylight Saving Time Started
+            // at 02:00:00 clocks were turned forward 0:30 hours to 02:30:00
+            assertInterval(time("1991-10-27T00:00:00.000+10:30"), time("1991-10-27T01:00:00.000+10:30"), rounding, 60, tz);
+            // the interval containing the switch time is 90 minutes long
+            assertInterval(time("1991-10-27T01:00:00.000+10:30"), time("1991-10-27T03:00:00.000+11:00"), rounding, 90, tz);
+            assertInterval(time("1991-10-27T03:00:00.000+11:00"), time("1991-10-27T04:00:00.000+11:00"), rounding, 60, tz);
+        }
+
+        {
+            // time zone "Pacific/Chatham"
+            // 5 Apr 2015 - Daylight Saving Time Ended
+            // at 03:45:00 clocks were turned backward 1 hour to 02:45:00
+            ZoneId tz = ZoneId.of("Pacific/Chatham");
+            Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.HOUR_OF_DAY, tz);
+
+            assertInterval(time("2015-04-05T02:00:00.000+13:45"), time("2015-04-05T03:00:00.000+13:45"), rounding, 60, tz);
+            assertInterval(time("2015-04-05T03:00:00.000+13:45"), time("2015-04-05T03:00:00.000+12:45"), rounding, 60, tz);
+            assertInterval(time("2015-04-05T03:00:00.000+12:45"), time("2015-04-05T04:00:00.000+12:45"), rounding, 60, tz);
+
+            // 27 Sep 2015 - Daylight Saving Time Started
+            // at 02:45:00 clocks were turned forward 1 hour to 03:45:00
+
+            assertInterval(time("2015-09-27T01:00:00.000+12:45"), time("2015-09-27T02:00:00.000+12:45"), rounding, 60, tz);
+            assertInterval(time("2015-09-27T02:00:00.000+12:45"), time("2015-09-27T04:00:00.000+13:45"), rounding, 60, tz);
+            assertInterval(time("2015-09-27T04:00:00.000+13:45"), time("2015-09-27T05:00:00.000+13:45"), rounding, 60, tz);
+        }
+    }
+
+    public void testDST_Europe_Rome() {
+        // time zone "Europe/Rome", rounding to days. Rome had two midnights on the day the clocks went back in 1978, and
+        // timeZone.convertLocalToUTC() gives the later of the two because Rome is east of UTC, whereas we want the earlier.
+
+        ZoneId tz = ZoneId.of("Europe/Rome");
+        Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.DAY_OF_MONTH, tz);
+
+        {
+            long timeBeforeFirstMidnight = time("1978-09-30T23:59:00+02:00");
+            long floor = rounding.round(timeBeforeFirstMidnight);
+            assertThat(floor, isDate(time("1978-09-30T00:00:00+02:00"), tz));
+        }
+
+        {
+            long timeBetweenMidnights = time("1978-10-01T00:30:00+02:00");
+            long floor = rounding.round(timeBetweenMidnights);
+            assertThat(floor, isDate(time("1978-10-01T00:00:00+02:00"), tz));
+        }
+
+        {
+            long timeAfterSecondMidnight = time("1978-10-01T00:30:00+01:00");
+            long floor = rounding.round(timeAfterSecondMidnight);
+            assertThat(floor, isDate(time("1978-10-01T00:00:00+02:00"), tz));
+
+            long prevFloor = rounding.round(floor - 1);
+            assertThat(prevFloor, lessThan(floor));
+            assertThat(prevFloor, isDate(time("1978-09-30T00:00:00+02:00"), tz));
+        }
+    }
+
+    /**
+     * Test for a time zone whose days overlap because the clocks are set back across midnight at the end of DST.
+     */
+    public void testDST_America_St_Johns() {
+        // time zone "America/St_Johns", rounding to days.
+        ZoneId tz = ZoneId.of("America/St_Johns");
+        Rounding rounding = new Rounding.TimeUnitRounding(Rounding.DateTimeUnit.DAY_OF_MONTH, tz);
+
+        // 29 October 2006 - Daylight Saving Time ended, changing the UTC offset from -02:30 to -03:30.
+        // This happened at 02:31 UTC, 00:01 local time, so the clocks were set back 1 hour to 23:01 on the 28th.
+        // This means that 2006-10-29 has _two_ midnights, one in the -02:30 offset and one in the -03:30 offset.
+        // Only the first of these is considered "rounded". Moreover, the extra time between 23:01 and 23:59
+        // should be considered as part of the 28th even though it comes after midnight on the 29th.
+
+        {
+            // Times before the first midnight should be rounded up to the first midnight.
+            long timeBeforeFirstMidnight = time("2006-10-28T23:30:00.000-02:30");
+            long floor = rounding.round(timeBeforeFirstMidnight);
+            assertThat(floor, isDate(time("2006-10-28T00:00:00.000-02:30"), tz));
+            long ceiling = rounding.nextRoundingValue(timeBeforeFirstMidnight);
+            assertThat(ceiling, isDate(time("2006-10-29T00:00:00.000-02:30"), tz));
+            assertInterval(floor, timeBeforeFirstMidnight, ceiling, rounding, tz);
+        }
+
+        {
+            // Times between the two midnights which are on the later day should be rounded down to the later day's midnight.
+            long timeBetweenMidnights = time("2006-10-29T00:00:30.000-02:30");
+            // (this is halfway through the last minute before the clocks changed, in which local time was ambiguous)
+
+            long floor = rounding.round(timeBetweenMidnights);
+            assertThat(floor, isDate(time("2006-10-29T00:00:00.000-02:30"), tz));
+
+            long ceiling = rounding.nextRoundingValue(timeBetweenMidnights);
+            assertThat(ceiling, isDate(time("2006-10-30T00:00:00.000-03:30"), tz));
+
+            assertInterval(floor, timeBetweenMidnights, ceiling, rounding, tz);
+        }
+
+        {
+            // Times between the two midnights which are on the earlier day should be rounded down to the earlier day's midnight.
+            long timeBetweenMidnights = time("2006-10-28T23:30:00.000-03:30");
+            // (this is halfway through the hour after the clocks changed, in which local time was ambiguous)
+
+            long floor = rounding.round(timeBetweenMidnights);
+            assertThat(floor, isDate(time("2006-10-28T00:00:00.000-02:30"), tz));
+
+            long ceiling = rounding.nextRoundingValue(timeBetweenMidnights);
+            assertThat(ceiling, isDate(time("2006-10-29T00:00:00.000-02:30"), tz));
+
+            assertInterval(floor, timeBetweenMidnights, ceiling, rounding, tz);
+        }
+
+        {
+            // Times after the second midnight should be rounded down to the first midnight.
+            long timeAfterSecondMidnight = time("2006-10-29T06:00:00.000-03:30");
+            long floor = rounding.round(timeAfterSecondMidnight);
+            assertThat(floor, isDate(time("2006-10-29T00:00:00.000-02:30"), tz));
+            long ceiling = rounding.nextRoundingValue(timeAfterSecondMidnight);
+            assertThat(ceiling, isDate(time("2006-10-30T00:00:00.000-03:30"), tz));
+            assertInterval(floor, timeAfterSecondMidnight, ceiling, rounding, tz);
+        }
+    }
+
+    /**
+     * tests for dst transition with overlaps and day roundings.
+     */
+    public void testDST_END_Edgecases() {
+        // First case, dst happens at 1am local time, switching back one hour.
+        // We want the overlapping hour to count for the next day, making it a 25h interval
+
+        ZoneId tz = ZoneId.of("Atlantic/Azores");
+        Rounding.DateTimeUnit timeUnit = Rounding.DateTimeUnit.DAY_OF_MONTH;
+        Rounding rounding = new Rounding.TimeUnitRounding(timeUnit, tz);
+
+        // Sunday, 29 October 2000, 01:00:00 clocks were turned backward 1 hour
+        // to Sunday, 29 October 2000, 00:00:00 local standard time instead
+        // which means there were two midnights that day.
+
+        long midnightBeforeTransition = time("2000-10-29T00:00:00", tz);
+        long midnightOfTransition = time("2000-10-29T00:00:00-01:00");
+        assertEquals(60L * 60L * 1000L, midnightOfTransition - midnightBeforeTransition);
+        long nextMidnight = time("2000-10-30T00:00:00", tz);
+
+        assertInterval(midnightBeforeTransition, nextMidnight, rounding, 25 * 60, tz);
+
+        assertThat(rounding.round(time("2000-10-29T06:00:00-01:00")), isDate(time("2000-10-29T00:00:00Z"), tz));
+
+        // Second case, dst happens at 0am local time, switching back one hour to 23pm local time.
+        // We want the overlapping hour to count for the previous day here
+
+        tz = ZoneId.of("America/Lima");
+        rounding = new Rounding.TimeUnitRounding(timeUnit, tz);
+
+        // Sunday, 1 April 1990, 00:00:00 clocks were turned backward 1 hour to
+        // Saturday, 31 March 1990, 23:00:00 local standard time instead
+
+        midnightBeforeTransition = time("1990-03-31T00:00:00.000-04:00");
+        nextMidnight = time("1990-04-01T00:00:00.000-05:00");
+        assertInterval(midnightBeforeTransition, nextMidnight, rounding, 25 * 60, tz);
+
+        // make sure the next interval is 24h long again
+        long midnightAfterTransition = time("1990-04-01T00:00:00.000-05:00");
+        nextMidnight = time("1990-04-02T00:00:00.000-05:00");
+        assertInterval(midnightAfterTransition, nextMidnight, rounding, 24 * 60, tz);
+    }
+
+    private void assertInterval(long rounded, long nextRoundingValue, Rounding rounding, int minutes,
+                                ZoneId tz) {
+        assertInterval(rounded, dateBetween(rounded, nextRoundingValue), nextRoundingValue, rounding, tz);
+        long millisPerMinute = 60_000;
+        assertEquals(millisPerMinute * minutes, nextRoundingValue - rounded);
+    }
+
+    /**
+     * perform a number on assertions and checks on {@link org.elasticsearch.common.rounding.Rounding.TimeUnitRounding} intervals
+     * @param rounded the expected low end of the rounding interval
+     * @param unrounded a date in the interval to be checked for rounding
+     * @param nextRoundingValue the expected upper end of the rounding interval
+     * @param rounding the rounding instance
+     */
+    private void assertInterval(long rounded, long unrounded, long nextRoundingValue, Rounding rounding, ZoneId tz) {
+        assertThat("rounding should be idempotent ", rounding.round(rounded), isDate(rounded, tz));
+        assertThat("rounded value smaller or equal than unrounded" + rounding, rounded, lessThanOrEqualTo(unrounded));
+        assertThat("values less than rounded should round further down" + rounding, rounding.round(rounded - 1), lessThan(rounded));
+        assertThat("nextRounding value should be a rounded date", rounding.round(nextRoundingValue), isDate(nextRoundingValue, tz));
+        assertThat("values above nextRounding should round down there", rounding.round(nextRoundingValue + 1),
+            isDate(nextRoundingValue, tz));
+
+        if (isTimeWithWellDefinedRounding(tz, unrounded)) {
+            assertThat("nextRounding value should be greater than date" + rounding, nextRoundingValue, greaterThan(unrounded));
+
+            long dateBetween = dateBetween(rounded, nextRoundingValue);
+            long roundingDateBetween = rounding.round(dateBetween);
+            ZonedDateTime zonedDateBetween = ZonedDateTime.ofInstant(Instant.ofEpochMilli(dateBetween), tz);
+            assertThat("dateBetween [" + zonedDateBetween + "/" + dateBetween + "] should round down to roundedDate [" +
+                    Instant.ofEpochMilli(roundingDateBetween) + "]", roundingDateBetween, isDate(rounded, tz));
+            assertThat("dateBetween [" + zonedDateBetween + "] should round up to nextRoundingValue",
+                rounding.nextRoundingValue(dateBetween), isDate(nextRoundingValue, tz));
+        }
+    }
+
+    private static boolean isTimeWithWellDefinedRounding(ZoneId tz, long t) {
+        if (tz.getId().equals("America/St_Johns")
+            || tz.getId().equals("America/Goose_Bay")
+            || tz.getId().equals("America/Moncton")
+            || tz.getId().equals("Canada/Newfoundland")) {
+
+            // Clocks went back at 00:01 between 1987 and 2010, causing overlapping days.
+            // These timezones are otherwise uninteresting, so just skip this period.
+
+            return t <= time("1987-10-01T00:00:00Z")
+                || t >= time("2010-12-01T00:00:00Z");
+        }
+
+        if (tz.getId().equals("Antarctica/Casey")) {
+
+            // Clocks went back 3 hours at 02:00 on 2010-03-05, causing overlapping days.
+
+            return t <= time("2010-03-03T00:00:00Z")
+                || t >= time("2010-03-07T00:00:00Z");
+        }
+
+        return true;
+    }
+
+    private static long dateBetween(long lower, long upper) {
+        long dateBetween = randomLongBetween(lower, upper - 1);
+        assert lower <= dateBetween && dateBetween < upper;
+        return dateBetween;
+    }
+
+    private static long time(String time) {
+        return time(time, ZoneOffset.UTC);
+    }
+
+    private static long time(String time, ZoneId zone) {
+        TemporalAccessor accessor = DateFormatters.forPattern("date_optional_time").withZone(zone).parse(time);
+        return DateFormatters.toZonedDateTime(accessor).toInstant().toEpochMilli();
+    }
+
+    private static Matcher<Long> isDate(final long expected, ZoneId tz) {
+        return new TypeSafeMatcher<Long>() {
+            @Override
+            public boolean matchesSafely(final Long item) {
+                return expected == item;
+            }
+
+            @Override
+            public void describeTo(Description description) {
+                ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(expected), tz);
+                description.appendText(DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(zonedDateTime) + " [" + expected + "] ");
+            }
+
+            @Override
+            protected void describeMismatchSafely(final Long actual, final Description mismatchDescription) {
+                ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(actual), tz);
+                mismatchDescription.appendText(" was ")
+                    .appendValue(DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(zonedDateTime) + " [" + actual + "]");
+            }
+        };
+    }
+
+}