BKLogReadHandler extends BKLogHandler { synchronized Future<Void> lockStream() { if (null == lockAcquireFuture) { final Function0<DistributedLock> lockFunction = new ExceptionalFunction0<DistributedLock>() { @Override public DistributedLock applyE() throws IOException { BKLogReadHandler.this.readLock = new ZKDistributedLock( lockStateExecutor, lockFactory, readLockPath, conf.getLockTimeoutMilliSeconds(), statsLogger.scope("read_lock")); LOG.info("acquiring readlock {} at {}", getLockClientId(), readLockPath); return BKLogReadHandler.this.readLock; } }; lockAcquireFuture = ensureReadLockPathExist().flatMap(new ExceptionalFunction<Void, Future<Void>>() { @Override public Future<Void> applyE(Void in) throws Throwable { return scheduler.apply(lockFunction).flatMap(new ExceptionalFunction<DistributedLock, Future<Void>>() { @Override public Future<Void> applyE(DistributedLock lock) throws IOException { return acquireLockOnExecutorThread(lock); } }); } }); } return lockAcquireFuture; } BKLogReadHandler(ZKLogMetadataForReader logMetadata,
                            Optional<String> subscriberId,
                            DistributedLogConfiguration conf,
                            DynamicDistributedLogConfiguration dynConf,
                            ZooKeeperClientBuilder zkcBuilder,
                            BookKeeperClientBuilder bkcBuilder,
                            LogSegmentMetadataStore metadataStore,
                            OrderedScheduler scheduler,
                            OrderedScheduler lockStateExecutor,
                            OrderedScheduler readAheadExecutor,
                            AlertStatsLogger alertStatsLogger,
                            ReadAheadExceptionsLogger readAheadExceptionsLogger,
                            StatsLogger statsLogger,
                            StatsLogger perLogStatsLogger,
                            String clientId,
                            AsyncNotification notification,
                            boolean isHandleForReading,
                            boolean deserializeRecordSet); Future<Void> asyncClose(); @Override Future<Void> asyncAbort(); void startReadAhead(LedgerReadPosition startPosition,
                               AsyncFailureInjector failureInjector); boolean isReadAheadCaughtUp(); LedgerHandleCache getHandleCache(); LogRecordWithDLSN getNextReadAheadRecord(); ReadAheadCache getReadAheadCache();  }
@Test(timeout = 60000) public void testLockStreamWithMissingLog() throws Exception { String streamName = runtime.getMethodName(); BKDistributedLogManager bkdlm = (BKDistributedLogManager) createNewDLM(conf, streamName); BKLogReadHandler readHandler = bkdlm.createReadHandler(); try { Await.result(readHandler.lockStream()); fail("Should fail lock stream if log not found"); } catch (LogNotFoundException ex) { } BKLogReadHandler subscriberReadHandler = bkdlm.createReadHandler(Optional.of("test-subscriber")); try { Await.result(subscriberReadHandler.lockStream()); fail("Subscriber should fail lock stream if log not found"); } catch (LogNotFoundException ex) { } }