@VisibleForTesting
  protected void setInitialFlushTime(Date now) {
    // Start with the beginning of the current hour
    nextFlush = Calendar.getInstance();
    nextFlush.setTime(now);
    nextFlush.set(Calendar.MILLISECOND, 0);
    nextFlush.set(Calendar.SECOND, 0);
    nextFlush.set(Calendar.MINUTE, 0);

    // In the first round, calculate the first flush as the largest number of
    // intervals from the beginning of the current hour that's not in the
    // future by:
    // 1. Subtract the beginning of the hour from the current time
    // 2. Divide by the roll interval and round down to get the number of whole
    //    intervals that have passed since the beginning of the hour
    // 3. Multiply by the roll interval to get the number of millis between
    //    the beginning of the current hour and the beginning of the current
    //    interval.
    int millis = (int) (((now.getTime() - nextFlush.getTimeInMillis())
        / rollIntervalMillis) * rollIntervalMillis);

    // Then add some noise to help prevent all the nodes from
    // closing their files at the same time.
    if (rollOffsetIntervalMillis > 0) {
      millis += ThreadLocalRandom.current().nextLong(rollOffsetIntervalMillis);

      // If the added time puts us into the future, step back one roll interval
      // because the code to increment nextFlush to the next flush expects that
      // nextFlush is the next flush from the previous interval.  There wasn't
      // a previous interval, so we just fake it with the time in the past that
      // would have been the previous interval if there had been one.
      //
      // It's OK if millis comes out negative.
      while (nextFlush.getTimeInMillis() + millis > now.getTime()) {
        millis -= rollIntervalMillis;
      }
    }

    // Adjust the next flush time by millis to get the time of our ficticious
    // previous next flush
    nextFlush.add(Calendar.MILLISECOND, millis);
  }
@Test
  public void testSetInitialFlushTime() {
    RollingFileSystemSink rfsSink = new RollingFileSystemSink(1000, 0);
    Calendar calendar = Calendar.getInstance();

    calendar.set(Calendar.MILLISECOND, 0);
    calendar.set(Calendar.SECOND, 0);
    calendar.set(Calendar.MINUTE, 0);
    calendar.set(Calendar.HOUR, 0);
    calendar.set(Calendar.DAY_OF_YEAR, 1);
    calendar.set(Calendar.YEAR, 2016);

    assertNull("Last flush time should have been null prior to calling init()",
        rfsSink.nextFlush);

    rfsSink.setInitialFlushTime(calendar.getTime());

    long diff =
        rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertEquals("The initial flush time was calculated incorrectly", 0L, diff);

    calendar.set(Calendar.MILLISECOND, 10);
    rfsSink.setInitialFlushTime(calendar.getTime());
    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertEquals("The initial flush time was calculated incorrectly",
        -10L, diff);

    calendar.set(Calendar.SECOND, 1);
    calendar.set(Calendar.MILLISECOND, 10);
    rfsSink.setInitialFlushTime(calendar.getTime());
    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertEquals("The initial flush time was calculated incorrectly",
        -10L, diff);

    // Try again with a random offset
    rfsSink = new RollingFileSystemSink(1000, 100);

    assertNull("Last flush time should have been null prior to calling init()",
        rfsSink.nextFlush);

    calendar.set(Calendar.MILLISECOND, 0);
    calendar.set(Calendar.SECOND, 0);
    rfsSink.setInitialFlushTime(calendar.getTime());

    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertTrue("The initial flush time was calculated incorrectly: " + diff,
        (diff == 0L) || ((diff > -1000L) && (diff < -900L)));

    calendar.set(Calendar.MILLISECOND, 10);
    rfsSink.setInitialFlushTime(calendar.getTime());
    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertTrue("The initial flush time was calculated incorrectly: " + diff,
        (diff >= -10L) && (diff <= 0L) || ((diff > -1000L) && (diff < -910L)));

    calendar.set(Calendar.SECOND, 1);
    calendar.set(Calendar.MILLISECOND, 10);
    rfsSink.setInitialFlushTime(calendar.getTime());
    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertTrue("The initial flush time was calculated incorrectly: " + diff,
        (diff >= -10L) && (diff <= 0L) || ((diff > -1000L) && (diff < -910L)));

    // Now try pathological settings
    rfsSink = new RollingFileSystemSink(1000, 1000000);

    assertNull("Last flush time should have been null prior to calling init()",
        rfsSink.nextFlush);

    calendar.set(Calendar.MILLISECOND, 1);
    calendar.set(Calendar.SECOND, 0);
    rfsSink.setInitialFlushTime(calendar.getTime());

    diff = rfsSink.nextFlush.getTimeInMillis() - calendar.getTimeInMillis();

    assertTrue("The initial flush time was calculated incorrectly: " + diff,
        (diff > -1000L) && (diff <= 0L));
  }