BucketWriter { @SuppressWarnings({"squid:S1860","squid:MethodCyclomaticComplexity","squid:S134","squid:S1160"}) 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) { LOGGER.warn("Timeout while trying to cancel closing of idle file. Idle" + " file close may have failed", ex); } catch (Exception ex) { LOGGER.warn("Error while trying to cancel closing of idle file. ", ex); } } idleFuture = null; } if (!isOpen) { if (closed) { 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 (MAX_CONSEC_UNDER_REPL_ROTATIONS > 0 && consecutiveUnderReplRotateCount >= MAX_CONSEC_UNDER_REPL_ROTATIONS) { doRotate = false; if (consecutiveUnderReplRotateCount == MAX_CONSEC_UNDER_REPL_ROTATIONS) { LOGGER.error("Hit max consecutive under-replication rotations ({}); " + "will not continue rolling files under this path due to " + "under-replication", MAX_CONSEC_UNDER_REPL_ROTATIONS); } } else { LOGGER.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) { LOGGER.warn("Caught IOException writing to HDFSWriter ({}). Closing file (" + bucketPath + ") and rethrowing exception.", e.getMessage()); try { close(true); } catch (IOException e2) { LOGGER.warn("Caught IOException while closing file (" + bucketPath + "). Exception follows.", e2); } throw e; } processSize += event.getBody().length; eventCounter++; batchCounter++; if (batchCounter == batchSize) { flush(); } } @SuppressWarnings("squid:S00107")  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 maxCloseTries,String processId); @SuppressWarnings("squid:S1160") synchronized void close(); @SuppressWarnings({"squid:S1160","squid:MethodCyclomaticComplexity", "squid:S1192"}) synchronized void close(boolean callCloseCallback); @SuppressWarnings("squid:S1160") synchronized void flush(); @SuppressWarnings({"squid:S1860","squid:MethodCyclomaticComplexity","squid:S134","squid:S1160"}) synchronized void append(final Event event); @Override String toString();  }
@Test @Ignore public void testEventCountingRoller() throws IOException, InterruptedException { int maxEvents = 100; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); BucketWriter bucketWriter = new BucketWriter(0, 0, maxEvents, 0, ctx, "/tmp", "file", "", ".tmp", null, null, SequenceFile.CompressionType.NONE, hdfsWriter, timedRollerPool, proxy, new SinkCounter("test-bucket-writer-" + System.currentTimeMillis()), 0, null, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); 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 @Ignore public void testSizeRoller() throws IOException, InterruptedException { int maxBytes = 300; MockHDFSWriter hdfsWriter = new MockHDFSWriter(); BucketWriter bucketWriter = new BucketWriter(0, maxBytes, 0, 0, ctx, "/tmp", "file", "", ".tmp", null, null, SequenceFile.CompressionType.NONE, hdfsWriter,timedRollerPool, proxy, new SinkCounter("test-bucket-writer-" + System.currentTimeMillis()),0, null, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); 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 @Ignore 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 BucketWriter(ROLL_INTERVAL, 0, 0, 0, ctx, "/tmp", "file", "", ".tmp", null, null, SequenceFile.CompressionType.NONE, hdfsWriter, timedRollerPool, proxy, new SinkCounter("test-bucket-writer-" + System.currentTimeMillis()), 0, new BDREHDFSEventSink.WriterCallback() { @Override public void run(String filePath) { calledBack.set(true); } }, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); 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); Assert.assertTrue(calledBack.get()); bucketWriter = new BucketWriter(ROLL_INTERVAL, 0, 0, 0, ctx, "/tmp", "file", "", ".tmp", null, null, SequenceFile.CompressionType.NONE, hdfsWriter, timedRollerPool, proxy, new SinkCounter("test-bucket-writer-" + System.currentTimeMillis()), 0, null, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); 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 @Ignore 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 BucketWriter(ROLL_INTERVAL, 0, 0, 0, ctx, "/tmp", "file", PREFIX, ".tmp", null, null, SequenceFile.CompressionType.NONE, hdfsWriter, timedRollerPool, proxy, new SinkCounter( "test-bucket-writer-" + System.currentTimeMillis()), 0, null, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); bucketWriter.append(e); Assert.assertTrue("Incorrect in use prefix", hdfsWriter.getOpenedFilePath().contains(PREFIX)); }
@Test @Ignore 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 BucketWriter(ROLL_INTERVAL, 0, 0, 0, ctx, "/tmp", "file", "", SUFFIX, null, null, SequenceFile.CompressionType.NONE, hdfsWriter, timedRollerPool, proxy, new SinkCounter( "test-bucket-writer-" + System.currentTimeMillis()), 0, null, null, 30000, Executors.newSingleThreadExecutor(), 0, 0, "1"); Event e = EventBuilder.withBody("foo", Charsets.UTF_8); bucketWriter.append(e); Assert.assertTrue("Incorrect in use suffix", hdfsWriter.getOpenedFilePath().contains(SUFFIX)); }