@Test
public void testIntervalRoller() throws IOException, InterruptedException {

  final int ROLL_INTERVAL = 1; // seconds
  final int NUM_EVENTS = 10;
  final AtomicBoolean calledBack = new AtomicBoolean(false);
  MockHDFSWriter hdfsWriter = new MockHDFSWriter();
  BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter)
      .setRollInterval(ROLL_INTERVAL)
      .setOnCloseCallback(new HDFSEventSink.WriterCallback() {
        @Override
        public void run(String filePath) {
          calledBack.set(true);
        }
      }).build();
  Event e = EventBuilder.withBody("foo", Charsets.UTF_8);
  long startNanos = System.nanoTime();
  for (int i = 0; i < NUM_EVENTS - 1; i++) {
    bucketWriter.append(e);
  }
  // sleep to force a roll... wait 2x interval just to be sure
  Thread.sleep(2 * ROLL_INTERVAL * 1000L);
  Assert.assertTrue(bucketWriter.closed.get());
  Assert.assertTrue(calledBack.get());
  bucketWriter = new BucketWriterBuilder(hdfsWriter)
      .setRollInterval(ROLL_INTERVAL).build();
  // write one more event (to reopen a new file so we will roll again later)
  bucketWriter.append(e);
  long elapsedMillis = TimeUnit.MILLISECONDS.convert(
      System.nanoTime() - startNanos, TimeUnit.NANOSECONDS);
  long elapsedSeconds = elapsedMillis / 1000L;
  logger.info("Time elapsed: {} milliseconds", elapsedMillis);
  logger.info("Number of events written: {}", hdfsWriter.getEventsWritten());
  logger.info("Number of bytes written: {}", hdfsWriter.getBytesWritten());
  logger.info("Number of files opened: {}", hdfsWriter.getFilesOpened());
  logger.info("Number of files closed: {}", hdfsWriter.getFilesClosed());
  Assert.assertEquals("events written", NUM_EVENTS,
      hdfsWriter.getEventsWritten());
  Assert.assertEquals("bytes written", e.getBody().length * NUM_EVENTS,
      hdfsWriter.getBytesWritten());
  Assert.assertEquals("files opened", 2, hdfsWriter.getFilesOpened());
  // before auto-roll
  Assert.assertEquals("files closed", 1, hdfsWriter.getFilesClosed());
  logger.info("Waiting for roll...");
  Thread.sleep(2 * ROLL_INTERVAL * 1000L);
  logger.info("Number of files closed: {}", hdfsWriter.getFilesClosed());
  Assert.assertEquals("files closed", 2, hdfsWriter.getFilesClosed());
}