RollingFileSystemSink implements MetricsSink, Closeable { @VisibleForTesting protected void setInitialFlushTime(Date now) { nextFlush = Calendar.getInstance(); nextFlush.setTime(now); nextFlush.set(Calendar.MILLISECOND, 0); nextFlush.set(Calendar.SECOND, 0); nextFlush.set(Calendar.MINUTE, 0); int millis = (int) (((now.getTime() - nextFlush.getTimeInMillis()) / rollIntervalMillis) * rollIntervalMillis); if (rollOffsetIntervalMillis > 0) { millis += ThreadLocalRandom.current().nextLong(rollOffsetIntervalMillis); while (nextFlush.getTimeInMillis() + millis > now.getTime()) { millis -= rollIntervalMillis; } } nextFlush.add(Calendar.MILLISECOND, millis); } RollingFileSystemSink(); @VisibleForTesting protected  RollingFileSystemSink(long flushIntervalMillis,
      long flushOffsetIntervalMillis); @Override void init(SubsetConfiguration metrics2Properties); @Override void putMetrics(MetricsRecord record); @Override void flush(); @Override void close();  }
@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); 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))); 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)); }