BucketWriter { public synchronized void append(final Event event) throws IOException, InterruptedException { checkAndThrowInterruptedException(); if (idleFuture != null) { idleFuture.cancel(false); if (!idleFuture.isDone()) { try { idleFuture.get(callTimeout, TimeUnit.MILLISECONDS); } catch (TimeoutException ex) { LOG.warn("Timeout while trying to cancel closing of idle file. Idle" + " file close may have failed", ex); } catch (Exception ex) { LOG.warn("Error while trying to cancel closing of idle file. ", ex); } } idleFuture = null; } if (!isOpen) { if (closed.get()) { throw new BucketClosedException("This bucket writer was closed and " + "this handle is thus no longer valid"); } open(); } if (shouldRotate()) { boolean doRotate = true; if (isUnderReplicated) { if (maxConsecUnderReplRotations > 0 && consecutiveUnderReplRotateCount >= maxConsecUnderReplRotations) { doRotate = false; if (consecutiveUnderReplRotateCount == maxConsecUnderReplRotations) { LOG.error("Hit max consecutive under-replication rotations ({}); " + "will not continue rolling files under this path due to " + "under-replication", maxConsecUnderReplRotations); } } else { LOG.warn("Block Under-replication detected. Rotating file."); } consecutiveUnderReplRotateCount++; } else { consecutiveUnderReplRotateCount = 0; } if (doRotate) { close(); open(); } } try { sinkCounter.incrementEventDrainAttemptCount(); callWithTimeout(new CallRunner<Void>() { @Override public Void call() throws Exception { writer.append(event); return null; } }); } catch (IOException e) { LOG.warn("Caught IOException writing to HDFSWriter ({}). Closing file (" + bucketPath + ") and rethrowing exception.", e.getMessage()); close(true); throw e; } processSize += event.getBody().length; eventCounter++; batchCounter++; if (batchCounter == batchSize) { flush(); } } BucketWriter(long rollInterval, long rollSize, long rollCount, long batchSize,
      Context context, String filePath, String fileName, String inUsePrefix,
      String inUseSuffix, String fileSuffix, CompressionCodec codeC,
      CompressionType compType, HDFSWriter writer,
      ScheduledExecutorService timedRollerPool, PrivilegedExecutor proxyUser,
      SinkCounter sinkCounter, int idleTimeout, WriterCallback onCloseCallback,
      String onCloseCallbackPath, long callTimeout,
      ExecutorService callTimeoutPool, long retryInterval,
      int maxRetries);   BucketWriter(long rollInterval, long rollSize, long rollCount, long batchSize,
           Context context, String filePath, String fileName, String inUsePrefix,
           String inUseSuffix, String fileSuffix, CompressionCodec codeC,
           CompressionType compType, HDFSWriter writer,
           ScheduledExecutorService timedRollerPool, PrivilegedExecutor proxyUser,
           SinkCounter sinkCounter, int idleTimeout, WriterCallback onCloseCallback,
           String onCloseCallbackPath, long callTimeout,
           ExecutorService callTimeoutPool, long retryInterval,
           int maxRetries, Clock clock); void close(); void close(boolean callCloseCallback); void close(boolean callCloseCallback, boolean immediate); synchronized void flush(); synchronized void append(final Event event); @Override String toString();  }
@Test public void testEventCountingRoller() throws IOException, InterruptedException { int maxEvents = 100; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter) .setRollCount(maxEvents).build(); 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, InterruptedException { int maxBytes = 300; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter) .setRollSize(maxBytes).build(); 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 { final int ROLL_INTERVAL = 1; 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); } Thread.sleep(2 * ROLL_INTERVAL * 1000L); Assert.assertTrue(bucketWriter.closed.get()); Assert.assertTrue(calledBack.get()); bucketWriter = new BucketWriterBuilder(hdfsWriter) .setRollInterval(ROLL_INTERVAL).build(); 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()); 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()); }
@Test public void testInUsePrefix() throws IOException, InterruptedException { final int ROLL_INTERVAL = 1000; final String PREFIX = "BRNO_IS_CITY_IN_CZECH_REPUBLIC"; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); HDFSTextSerializer formatter = new HDFSTextSerializer(); BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter) .setRollInterval(ROLL_INTERVAL) .setInUsePrefix(PREFIX) .build(); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); bucketWriter.append(e); Assert.assertTrue("Incorrect in use prefix", hdfsWriter.getOpenedFilePath().contains(PREFIX)); }
@Test public void testInUseSuffix() throws IOException, InterruptedException { final int ROLL_INTERVAL = 1000; final String SUFFIX = "WELCOME_TO_THE_HELLMOUNTH"; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); HDFSTextSerializer serializer = new HDFSTextSerializer(); BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter) .setRollInterval(ROLL_INTERVAL) .setInUseSuffix(SUFFIX) .build(); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); bucketWriter.append(e); Assert.assertTrue("Incorrect in use suffix", hdfsWriter.getOpenedFilePath().contains(SUFFIX)); }
@Test public void testRotateBucketOnIOException() throws IOException, InterruptedException { MockHDFSWriter hdfsWriter = Mockito.spy(new MockHDFSWriter()); PrivilegedExecutor ugiProxy = FlumeAuthenticationUtil.getAuthenticator(null, null).proxyAs("alice"); final int ROLL_COUNT = 1; BucketWriter bucketWriter = new BucketWriterBuilder(hdfsWriter) .setProxyUser(ugiProxy) .setRollCount(ROLL_COUNT) .build(); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); bucketWriter.append(e); IOException expectedIOException = new IOException("Test injected IOException"); Mockito.doThrow(expectedIOException).when(hdfsWriter) .append(Mockito.any(Event.class)); try { bucketWriter.append(e); Assert.fail("Expected IOException wasn't thrown during append"); } catch (IOException ex) { Assert.assertEquals(expectedIOException, ex); logger.info("Caught expected IOException", ex); } try { bucketWriter.append(e); Assert.fail("BucketWriter should be already closed, BucketClosedException expected"); } catch (BucketClosedException ex) { logger.info("Caught expected BucketClosedException", ex); } Assert.assertEquals("events written", 1, hdfsWriter.getEventsWritten()); Assert.assertEquals("2 files should be closed", 2, hdfsWriter.getFilesClosed()); }