### Question:
BKSyncLogReader implements LogReader, AsyncNotification { @Override public synchronized LogRecordWithDLSN readNext(boolean nonBlocking) throws IOException { checkClosedOrException(); LogRecordWithDLSN record = doReadNext(nonBlocking); if (null == record && shouldCheckIdleReader) { if (readAheadReader.getNumCachedEntries() <= 0 && readAheadReader.isReaderIdle(idleErrorThresholdMillis, TimeUnit.MILLISECONDS)) { markReaderAsIdle(); } } return record; } BKSyncLogReader(DistributedLogConfiguration conf,
                    BKDistributedLogManager bkdlm,
                    DLSN startDLSN,
                    Optional<Long> startTransactionId,
                    StatsLogger statsLogger); @Override synchronized LogRecordWithDLSN readNext(boolean nonBlocking); @Override synchronized List<LogRecordWithDLSN> readBulk(boolean nonBlocking, int numLogRecords); @Override CompletableFuture<Void> asyncClose(); @Override void close(); @Override void notifyOnError(Throwable cause); @Override void notifyOnOperationComplete();  }### Answer:
@Test(timeout = 60000) public void testCreateReaderBeyondLastTransactionId() throws Exception { String name = testName.getMethodName(); DistributedLogManager dlm = createNewDLM(conf, name); BKSyncLogWriter out = (BKSyncLogWriter) dlm.startLogSegmentNonPartitioned(); for (long i = 1; i < 10; i++) { LogRecord op = DLMTestUtil.getLogRecordInstance(i); out.write(op); } out.closeAndComplete(); LogReader reader = dlm.getInputStream(20L); assertNull(reader.readNext(false)); out = (BKSyncLogWriter) dlm.startLogSegmentNonPartitioned(); for (long i = 10; i < 30; i++) { LogRecord op = DLMTestUtil.getLogRecordInstance(i); out.write(op); } out.closeAndComplete(); for (int i = 0; i < 10; i++) { LogRecord record = waitForNextRecord(reader); assertEquals(20L + i, record.getTransactionId()); } assertNull(reader.readNext(false)); }
@Test(timeout = 60000) public void testDeletingLogWhileReading() throws Exception { String name = testName.getMethodName(); DistributedLogManager dlm = createNewDLM(conf, name); BKSyncLogWriter out = (BKSyncLogWriter) dlm.startLogSegmentNonPartitioned(); for (long i = 1; i < 10; i++) { LogRecord op = DLMTestUtil.getLogRecordInstance(i); out.write(op); } out.closeAndComplete(); LogReader reader = dlm.getInputStream(1L); for (int i = 1; i < 10; i++) { LogRecord record = waitForNextRecord(reader); assertEquals((long) i, record.getTransactionId()); } DistributedLogManager deleteDLM = createNewDLM(conf, name); deleteDLM.delete(); LogRecord record; try { record = reader.readNext(false); while (null == record) { record = reader.readNext(false); } fail("Should fail reading next with LogNotFound"); } catch (LogNotFoundException lnfe) { } }