BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Sizable { @Override synchronized public Future<DLSN> asyncWrite(LogRecord record) { return asyncWrite(record, true); } protected  BKLogSegmentWriter(String streamName,
                                 String logSegmentName,
                                 DistributedLogConfiguration conf,
                                 int logSegmentMetadataVersion,
                                 LogSegmentEntryWriter entryWriter,
                                 DistributedLock lock, /** the lock needs to be acquired **/
                                 long startTxId,
                                 long logSegmentSequenceNumber,
                                 OrderedScheduler scheduler,
                                 StatsLogger statsLogger,
                                 StatsLogger perLogStatsLogger,
                                 AlertStatsLogger alertStatsLogger,
                                 PermitLimiter globalWriteLimiter,
                                 FeatureProvider featureProvider,
                                 DynamicDistributedLogConfiguration dynConf); @Override long getLogSegmentId(); @Override long size(); @Override Future<Void> asyncClose(); @Override Future<Void> asyncAbort(); @Override synchronized void write(LogRecord record); @Override synchronized Future<DLSN> asyncWrite(LogRecord record); synchronized Future<DLSN> asyncWrite(LogRecord record, boolean flush); synchronized Future<DLSN> writeInternal(LogRecord record); Future<Long> markEndOfStream(); synchronized int writeBulk(List<LogRecord> records); @Override synchronized Future<Long> flush(); @Override synchronized Future<Long> commit(); @Override void addComplete(final int rc, LedgerHandle handle,
                            final long entryId, final Object ctx); @Override synchronized void run();  }
@Test(timeout = 60000) public void testNondurableWriteAfterLedgerIsFenced() throws Exception { DistributedLogConfiguration confLocal = newLocalConf(); confLocal.setImmediateFlushEnabled(false); confLocal.setOutputBufferSize(Integer.MAX_VALUE); confLocal.setPeriodicFlushFrequencyMilliSeconds(0); confLocal.setDurableWriteEnabled(false); ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true); BKLogSegmentWriter writer = createLogSegmentWriter(confLocal, 0L, -1L, lock); fenceLedger(getLedgerHandle(writer)); LogRecord record = DLMTestUtil.getLogRecordInstance(1); record.setControl(); try { Await.result(writer.asyncWrite(record)); fail("Should fail the writer if the log segment is already fenced"); } catch (BKTransmitException bkte) { assertEquals(BKException.Code.LedgerFencedException, bkte.getBKResultCode()); } try { Await.result(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(2))); fail("Should fail the writer if the log segment is already fenced"); } catch (WriteException we) { } abortWriterAndLock(writer, lock); }