BKDistributedLogManager extends ZKMetadataAccessor implements DistributedLogManager { @Override public Future<AsyncLogWriter> openAsyncLogWriter() { try { checkClosedOrInError("startLogSegmentNonPartitioned"); } catch (AlreadyClosedException e) { return Future.exception(e); } Future<BKLogWriteHandler> createWriteHandleFuture; synchronized (this) { createWriteHandleFuture = asyncCreateWriteHandler(true); } return createWriteHandleFuture.flatMap(new AbstractFunction1<BKLogWriteHandler, Future<AsyncLogWriter>>() { @Override public Future<AsyncLogWriter> apply(final BKLogWriteHandler writeHandler) { final BKAsyncLogWriter writer; synchronized (BKDistributedLogManager.this) { writer = new BKAsyncLogWriter( conf, dynConf, BKDistributedLogManager.this, writeHandler, featureProvider, statsLogger); } return writeHandler.recoverIncompleteLogSegments() .map(new AbstractFunction1<Long, AsyncLogWriter>() { @Override public AsyncLogWriter apply(Long lastTxId) { writer.setLastTxId(lastTxId); return writer; } }).onFailure(new AbstractFunction1<Throwable, BoxedUnit>() { @Override public BoxedUnit apply(Throwable cause) { writer.asyncAbort(); return BoxedUnit.UNIT; } }); } }); } BKDistributedLogManager(String name,
                            DistributedLogConfiguration conf,
                            URI uri,
                            ZooKeeperClientBuilder writerZKCBuilder,
                            ZooKeeperClientBuilder readerZKCBuilder,
                            ZooKeeperClient zkcForWriterBKC,
                            ZooKeeperClient zkcForReaderBKC,
                            BookKeeperClientBuilder writerBKCBuilder,
                            BookKeeperClientBuilder readerBKCBuilder,
                            FeatureProvider featureProvider,
                            PermitLimiter writeLimiter,
                            StatsLogger statsLogger);   BKDistributedLogManager(String name,
                            DistributedLogConfiguration conf,
                            DynamicDistributedLogConfiguration dynConf,
                            URI uri,
                            ZooKeeperClientBuilder writerZKCBuilder,
                            ZooKeeperClientBuilder readerZKCBuilder,
                            ZooKeeperClient zkcForWriterBKC,
                            ZooKeeperClient zkcForReaderBKC,
                            BookKeeperClientBuilder writerBKCBuilder,
                            BookKeeperClientBuilder readerBKCBuilder,
                            SessionLockFactory lockFactory,
                            LogSegmentMetadataStore writerMetadataStore,
                            LogSegmentMetadataStore readerMetadataStore,
                            OrderedScheduler scheduler,
                            OrderedScheduler readAheadScheduler,
                            OrderedScheduler lockStateExecutor,
                            ClientSocketChannelFactory channelFactory,
                            HashedWheelTimer requestTimer,
                            ReadAheadExceptionsLogger readAheadExceptionsLogger,
                            String clientId,
                            Integer regionId,
                            LedgerAllocator ledgerAllocator,
                            PermitLimiter writeLimiter,
                            PermitManager logSegmentRollingPermitManager,
                            FeatureProvider featureProvider,
                            StatsLogger statsLogger,
                            StatsLogger perLogStatsLogger); @Override List<LogSegmentMetadata> getLogSegments(); @Override void registerListener(LogSegmentListener listener); @Override synchronized void unregisterListener(LogSegmentListener listener); void checkClosedOrInError(String operation); BKLogWriteHandler createWriteHandler(boolean lockHandler); @Override boolean isEndOfStreamMarked(); AppendOnlyStreamWriter getAppendOnlyStreamWriter(); AppendOnlyStreamReader getAppendOnlyStreamReader(); @Override BKSyncLogWriter startLogSegmentNonPartitioned(); @Override BKAsyncLogWriter startAsyncLogSegmentNonPartitioned(); @Override Future<AsyncLogWriter> openAsyncLogWriter(); @Override Future<DLSN> getDLSNNotLessThanTxId(final long fromTxnId); @Override LogReader getInputStream(long fromTxnId); @Override LogReader getInputStream(DLSN fromDLSN); @Override AsyncLogReader getAsyncLogReader(long fromTxnId); @Override Future<AsyncLogReader> openAsyncLogReader(long fromTxnId); @Override AsyncLogReader getAsyncLogReader(DLSN fromDLSN); @Override Future<AsyncLogReader> openAsyncLogReader(DLSN fromDLSN); @Override Future<AsyncLogReader> getAsyncLogReaderWithLock(final DLSN fromDLSN); @Override Future<AsyncLogReader> getAsyncLogReaderWithLock(final DLSN fromDLSN, final String subscriberId); @Override Future<AsyncLogReader> getAsyncLogReaderWithLock(String subscriberId); @Override LogRecordWithDLSN getLastLogRecord(); @Override long getFirstTxId(); @Override long getLastTxId(); @Override DLSN getLastDLSN(); @Override Future<LogRecordWithDLSN> getLastLogRecordAsync(); @Override Future<Long> getLastTxIdAsync(); @Override Future<DLSN> getFirstDLSNAsync(); @Override Future<DLSN> getLastDLSNAsync(); @Override long getLogRecordCount(); @Override Future<Long> getLogRecordCountAsync(final DLSN beginDLSN); @Override void recover(); @Override void delete(); @Override void purgeLogsOlderThan(long minTxIdToKeep); @Override Future<Void> asyncClose(); @Override void close(); boolean scheduleTask(Runnable task); @Override String toString(); void raiseAlert(String msg, Object... args); @Override @Deprecated SubscriptionStateStore getSubscriptionStateStore(String subscriberId); @Override SubscriptionsStore getSubscriptionsStore();  }
@Test(timeout = 60000) public void testTwoWritersOnLockDisabled() throws Exception { DistributedLogConfiguration confLocal = new DistributedLogConfiguration(); confLocal.addConfiguration(conf); confLocal.setOutputBufferSize(0); confLocal.setWriteLockEnabled(false); String name = "distrlog-two-writers-lock-disabled"; DistributedLogManager manager = createNewDLM(confLocal, name); AsyncLogWriter writer1 = FutureUtils.result(manager.openAsyncLogWriter()); FutureUtils.result(writer1.write(DLMTestUtil.getLogRecordInstance(1L))); AsyncLogWriter writer2 = FutureUtils.result(manager.openAsyncLogWriter()); FutureUtils.result(writer2.write(DLMTestUtil.getLogRecordInstance(2L))); try { FutureUtils.result(writer1.write(DLMTestUtil.getLogRecordInstance(3L))); fail("Should fail writing record to writer 1 again as writer 2 took over the ownership"); } catch (BKTransmitException bkte) { assertEquals(BKException.Code.LedgerFencedException, bkte.getBKResultCode()); } }