BucketWriter { public synchronized void append(Event event) throws IOException { if (!isOpen) { open(); } if (shouldRotate()) { close(); open(); } try { writer.append(event, formatter); } catch (IOException e) { LOG.warn("Caught IOException writing to HDFSWriter ({}). Closing file (" + bucketPath + IN_USE_EXT + ") and rethrowing exception.", e.getMessage()); try { close(); } catch (IOException e2) { LOG.warn("Caught IOException while closing file (" + bucketPath + IN_USE_EXT + "). Exception follows.", e2); } throw e; } processSize += event.getBody().length; eventCounter++; batchCounter++; if (batchCounter == batchSize) { flush(); } } BucketWriter(long rollInt, long rollSz, long rollCnt, long bSize,
      Context ctx, String fPath, CompressionCodec codec, CompressionType cType,
      HDFSWriter hWriter, FlumeFormatter fmt); synchronized void close(); synchronized void flush(); synchronized void append(Event event); @Override String toString(); boolean isBatchComplete();  }
@Test public void testEventCountingRoller() throws IOException { int maxEvents = 100; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); HDFSTextFormatter formatter = new HDFSTextFormatter(); BucketWriter bucketWriter = new BucketWriter(0, 0, maxEvents, 0, ctx, "/tmp/file", null, SequenceFile.CompressionType.NONE, hdfsWriter, formatter); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); for (int i = 0; i < 1000; i++) { bucketWriter.append(e); } logger.info("Number of events written: {}", hdfsWriter.getEventsWritten()); logger.info("Number of bytes written: {}", hdfsWriter.getBytesWritten()); logger.info("Number of files opened: {}", hdfsWriter.getFilesOpened()); Assert.assertEquals("events written", 1000, hdfsWriter.getEventsWritten()); Assert.assertEquals("bytes written", 3000, hdfsWriter.getBytesWritten()); Assert.assertEquals("files opened", 10, hdfsWriter.getFilesOpened()); }
@Test public void testSizeRoller() throws IOException { int maxBytes = 300; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); HDFSTextFormatter formatter = new HDFSTextFormatter(); BucketWriter bucketWriter = new BucketWriter(0, maxBytes, 0, 0, ctx, "/tmp/file", null, SequenceFile.CompressionType.NONE, hdfsWriter, formatter); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); for (int i = 0; i < 1000; i++) { bucketWriter.append(e); } logger.info("Number of events written: {}", hdfsWriter.getEventsWritten()); logger.info("Number of bytes written: {}", hdfsWriter.getBytesWritten()); logger.info("Number of files opened: {}", hdfsWriter.getFilesOpened()); Assert.assertEquals("events written", 1000, hdfsWriter.getEventsWritten()); Assert.assertEquals("bytes written", 3000, hdfsWriter.getBytesWritten()); Assert.assertEquals("files opened", 10, hdfsWriter.getFilesOpened()); }
@Test public void testIntervalRoller() throws IOException, InterruptedException { int rollInterval = 2; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); HDFSTextFormatter formatter = new HDFSTextFormatter(); BucketWriter bucketWriter = new BucketWriter(rollInterval, 0, 0, 0, ctx, "/tmp/file", null, SequenceFile.CompressionType.NONE, hdfsWriter, formatter); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); long startNanos = System.nanoTime(); for (int i = 0; i < 1000; i++) { bucketWriter.append(e); if (i % 100 == 0) { Thread.sleep(500L); } } 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()); Assert.assertEquals("events written", 1000, hdfsWriter.getEventsWritten()); Assert.assertEquals("bytes written", 3000, hdfsWriter.getBytesWritten()); Assert.assertEquals("files opened", elapsedSeconds/2 + 1, hdfsWriter.getFilesOpened()); }