BKDistributedLogManager extends ZKMetadataAccessor implements DistributedLogManager { @Override public DLSN getLastDLSN() throws IOException { checkClosedOrInError("getLastDLSN"); return FutureUtils.result(getLastLogRecordAsyncInternal(false, false)).getDlsn(); } 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 testGetLastDLSN() throws Exception { String name = "distrlog-get-last-dlsn"; DistributedLogConfiguration confLocal = new DistributedLogConfiguration(); confLocal.loadConf(conf); confLocal.setFirstNumEntriesPerReadLastRecordScan(2); confLocal.setMaxNumEntriesPerReadLastRecordScan(4); confLocal.setImmediateFlushEnabled(true); confLocal.setOutputBufferSize(0); DistributedLogManager dlm = createNewDLM(confLocal, name); BKAsyncLogWriter writer = (BKAsyncLogWriter) dlm.startAsyncLogSegmentNonPartitioned(); long txid = 1; LOG.info("Writing 10 control records"); for (int i = 0; i < 10; i++) { LogRecord record = DLMTestUtil.getLogRecordInstance(txid++); record.setControl(); Await.result(writer.writeControlRecord(record)); } LOG.info("10 control records are written"); try { dlm.getLastDLSN(); fail("Should fail on getting last dlsn from an empty log."); } catch (LogEmptyException lee) { } writer.closeAndComplete(); LOG.info("Completed first log segment"); writer = (BKAsyncLogWriter) dlm.startAsyncLogSegmentNonPartitioned(); Await.result(writer.write(DLMTestUtil.getLogRecordInstance(txid++))); LOG.info("Completed second log segment"); LOG.info("Writing another 10 control records"); for (int i = 1; i < 10; i++) { LogRecord record = DLMTestUtil.getLogRecordInstance(txid++); record.setControl(); Await.result(writer.write(record)); } assertEquals(new DLSN(2, 0, 0), dlm.getLastDLSN()); writer.closeAndComplete(); LOG.info("Completed third log segment"); assertEquals(new DLSN(2, 0, 0), dlm.getLastDLSN()); writer.close(); dlm.close(); }