Fluency implements Flushable, Closeable { public void emit(final String tag, final long timestamp, final Map<String, Object> data) throws IOException { emitter.emit(() -> buffer.append(tag, timestamp, data)); } Fluency(Buffer buffer, Flusher flusher); void emit(final String tag, final long timestamp, final Map<String, Object> data); void emit(String tag, Map<String, Object> data); void emit(final String tag, final EventTime eventTime, final Map<String, Object> data); void emit(final String tag, final long timestamp, final byte[] mapValue, final int offset, final int len); void emit(String tag, byte[] mapValue, int offset, int len); void emit(final String tag, final EventTime eventTime, final byte[] mapValue, final int offset, final int len); void emit(final String tag, final long timestamp, final ByteBuffer mapValue); void emit(String tag, ByteBuffer mapValue); void emit(final String tag, final EventTime eventTime, final ByteBuffer mapValue); @Override void flush(); @Override void close(); void clearBackupFiles(); long getAllocatedBufferSize(); long getBufferedDataSize(); boolean isTerminated(); boolean waitUntilAllBufferFlushed(int maxWaitSeconds); boolean waitUntilFlusherTerminated(int maxWaitSeconds); Buffer getBuffer(); Flusher getFlusher(); @Override String toString();  }
@Test public void testBufferFullException() throws IOException { final CountDownLatch latch = new CountDownLatch(1); StuckIngester stuckIngester = new StuckIngester(latch); bufferConfig.setChunkInitialSize(64); bufferConfig.setChunkExpandRatio(2); bufferConfig.setMaxBufferSize(256); bufferConfig.setChunkRetentionSize(196); flusherConfig.setFlushAttemptIntervalMillis(1000); Buffer buffer = new Buffer(bufferConfig, new JsonRecordFormatter()); Flusher flusher = new Flusher(flusherConfig, buffer, stuckIngester); try (Fluency fluency = new Fluency(buffer, flusher)) { Map<String, Object> event = new HashMap<>(); event.put("name", "xxxx"); for (int i = 0; i < 8; i++) { fluency.emit("tag", event); } try { fluency.emit("tag", event); fail(); } catch (BufferFullException e) { assertTrue(true); } finally { latch.countDown(); } } }