ReadUtils { public static Future<Optional<LogRecordWithDLSN>> getLogRecordNotLessThanTxId( final String logName, final LogSegmentMetadata segment, final long transactionId, final ExecutorService executorService, final LedgerHandleCache handleCache, final int nWays) { if (!segment.isInProgress()) { if (segment.getLastTxId() < transactionId) { Optional<LogRecordWithDLSN> noneRecord = Optional.absent(); return Future.value(noneRecord); } } final Promise<Optional<LogRecordWithDLSN>> promise = new Promise<Optional<LogRecordWithDLSN>>(); final FutureEventListener<LedgerDescriptor> openLedgerListener = new FutureEventListener<LedgerDescriptor>() { @Override public void onSuccess(final LedgerDescriptor ld) { promise.ensure(new AbstractFunction0<BoxedUnit>() { @Override public BoxedUnit apply() { handleCache.asyncCloseLedger(ld); return BoxedUnit.UNIT; } }); long lastEntryId; try { lastEntryId = handleCache.getLastAddConfirmed(ld); } catch (BKException e) { promise.setException(e); return; } if (lastEntryId < 0) { Optional<LogRecordWithDLSN> nonRecord = Optional.absent(); promise.setValue(nonRecord); return; } if (segment.getFirstTxId() >= transactionId) { final FirstTxIdNotLessThanSelector selector = new FirstTxIdNotLessThanSelector(transactionId); asyncReadRecordFromEntries( logName, ld, handleCache, segment, executorService, new SingleEntryScanContext(0L), selector ).addEventListener(new FutureEventListener<LogRecordWithDLSN>() { @Override public void onSuccess(LogRecordWithDLSN value) { promise.setValue(Optional.of(selector.result())); } @Override public void onFailure(Throwable cause) { promise.setException(cause); } }); return; } getLogRecordNotLessThanTxIdFromEntries( logName, ld, segment, transactionId, executorService, handleCache, Lists.newArrayList(0L, lastEntryId), nWays, Optional.<LogRecordWithDLSN>absent(), promise); } @Override public void onFailure(final Throwable cause) { String errMsg = "Error opening log segment [" + segment + "] for find record from " + logName; promise.setException(new IOException(errMsg, BKException.create(FutureUtils.bkResultCode(cause)))); } }; handleCache.asyncOpenLedger(segment, false) .addEventListener(FutureEventListenerRunnable.of(openLedgerListener, executorService)); return promise; }  static Future<LogRecordWithDLSN> asyncReadLastRecord(
            final String streamName,
            final LogSegmentMetadata l,
            final boolean fence,
            final boolean includeControl,
            final boolean includeEndOfStream,
            final int scanStartBatchSize,
            final int scanMaxBatchSize,
            final AtomicInteger numRecordsScanned,
            final ExecutorService executorService,
            final LedgerHandleCache handleCache); static Future<LogRecordWithDLSN> asyncReadFirstUserRecord(
            final String streamName,
            final LogSegmentMetadata l,
            final int scanStartBatchSize,
            final int scanMaxBatchSize,
            final AtomicInteger numRecordsScanned,
            final ExecutorService executorService,
            final LedgerHandleCache handleCache,
            final DLSN dlsn); static Future<Optional<LogRecordWithDLSN>> getLogRecordNotLessThanTxId(
            final String logName,
            final LogSegmentMetadata segment,
            final long transactionId,
            final ExecutorService executorService,
            final LedgerHandleCache handleCache,
            final int nWays);  }
@Test(timeout = 60000) public void testGetLogRecordNotLessThanTxIdWithGreaterTxId() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); DLMTestUtil.generateLogSegmentNonPartitioned(bkdlm, 0 , 1, 1 ); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, 999L)); assertFalse(result.isPresent()); }
@Test(timeout = 60000) public void testGetLogRecordNotLessThanTxIdWithLessTxId() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); DLMTestUtil.generateLogSegmentNonPartitioned(bkdlm, 0 , 1, 999L ); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, 99L)); assertTrue(result.isPresent()); assertEquals(999L, result.get().getTransactionId()); assertEquals(0L, result.get().getDlsn().getEntryId()); assertEquals(0L, result.get().getDlsn().getSlotId()); }
@Test(timeout = 60000) public void testGetLogRecordNotLessThanTxIdOnSmallSegment() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); DLMTestUtil.generateLogSegmentNonPartitioned(bkdlm, 0 , 5, 1L ); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, 3L)); assertTrue(result.isPresent()); assertEquals(3L, result.get().getTransactionId()); }
@Test(timeout = 60000) public void testGetLogRecordNotLessThanTxIdOnLargeSegment() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); DLMTestUtil.generateLogSegmentNonPartitioned(bkdlm, 0 , 100, 1L ); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, 9L)); assertTrue(result.isPresent()); assertEquals(9L, result.get().getTransactionId()); }
@Test(timeout = 60000) public void testGetLogRecordGreaterThanTxIdOnLargeSegment() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); DLMTestUtil.generateLogSegmentNonPartitioned(bkdlm, 0 , 100, 1L , 3L); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, 23L)); assertTrue(result.isPresent()); assertEquals(25L, result.get().getTransactionId()); }
@Test(timeout = 60000) public void testGetLogRecordGreaterThanTxIdOnSameTxId() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = createNewDLM(conf, streamName); AsyncLogWriter out = bkdlm.startAsyncLogSegmentNonPartitioned(); long txid = 1L; for (int i = 0; i < 10; ++i) { LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txid); Await.result(out.write(record)); txid += 1; } long txidToSearch = txid; for (int i = 0; i < 10; ++i) { LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txidToSearch); Await.result(out.write(record)); } for (int i = 0; i < 10; ++i) { LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txid); Await.result(out.write(record)); txid += 1; } Utils.close(out); Optional<LogRecordWithDLSN> result = FutureUtils.result(getLogRecordNotLessThanTxId(bkdlm, 0, txidToSearch)); assertTrue(result.isPresent()); assertEquals(10L, result.get().getDlsn().getEntryId()); }