/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.distributedlog;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import org.apache.distributedlog.callback.LogSegmentListener;
import org.apache.distributedlog.config.DynamicDistributedLogConfiguration;
import org.apache.distributedlog.exceptions.AlreadyClosedException;
import org.apache.distributedlog.exceptions.LogEmptyException;
import org.apache.distributedlog.exceptions.LogNotFoundException;
import org.apache.distributedlog.exceptions.UnexpectedException;
import org.apache.distributedlog.function.CloseAsyncCloseableFunction;
import org.apache.distributedlog.function.GetVersionedValueFunction;
import org.apache.distributedlog.injector.AsyncFailureInjector;
import org.apache.distributedlog.logsegment.LogSegmentEntryStore;
import org.apache.distributedlog.logsegment.LogSegmentEntryWriter;
import org.apache.distributedlog.metadata.LogMetadataForReader;
import org.apache.distributedlog.metadata.LogMetadataForWriter;
import org.apache.distributedlog.io.AsyncCloseable;
import org.apache.distributedlog.lock.DistributedLock;
import org.apache.distributedlog.lock.NopDistributedLock;
import org.apache.distributedlog.lock.ZKDistributedLock;
import org.apache.distributedlog.logsegment.LogSegmentFilter;
import org.apache.distributedlog.logsegment.LogSegmentMetadataCache;
import org.apache.distributedlog.metadata.LogStreamMetadataStore;
import org.apache.distributedlog.namespace.NamespaceDriver;
import org.apache.distributedlog.stats.BroadCastStatsLogger;
import org.apache.distributedlog.subscription.SubscriptionsStore;
import org.apache.distributedlog.util.Allocator;
import org.apache.distributedlog.util.DLUtils;
import org.apache.distributedlog.util.FutureUtils;
import org.apache.distributedlog.util.MonitoredFuturePool;
import org.apache.distributedlog.util.OrderedScheduler;
import org.apache.distributedlog.util.PermitLimiter;
import org.apache.distributedlog.util.PermitManager;
import org.apache.distributedlog.util.SchedulerUtils;
import org.apache.distributedlog.util.Utils;
import com.twitter.util.ExceptionalFunction;
import com.twitter.util.ExceptionalFunction0;
import com.twitter.util.Function;
import com.twitter.util.Future;
import com.twitter.util.FutureEventListener;
import com.twitter.util.Promise;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.runtime.AbstractFunction0;
import scala.runtime.AbstractFunction1;
import scala.runtime.BoxedUnit;

import java.io.Closeable;
import java.io.IOException;
import java.net.URI;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import static org.apache.distributedlog.namespace.NamespaceDriver.Role.READER;
import static org.apache.distributedlog.namespace.NamespaceDriver.Role.WRITER;

/**
 * <h3>Metrics</h3>
 * <ul>
 * <li> `log_writer/*`: all asynchronous writer related metrics are exposed under scope `log_writer`.
 * See {@link BKAsyncLogWriter} for detail stats.
 * <li> `async_reader/*`: all asyncrhonous reader related metrics are exposed under scope `async_reader`.
 * See {@link BKAsyncLogReader} for detail stats.
 * <li> `writer_future_pool/*`: metrics about the future pools that used by writers are exposed under
 * scope `writer_future_pool`. See {@link MonitoredFuturePool} for detail stats.
 * <li> `reader_future_pool/*`: metrics about the future pools that used by readers are exposed under
 * scope `reader_future_pool`. See {@link MonitoredFuturePool} for detail stats.
 * <li> `lock/*`: metrics about the locks used by writers. See {@link ZKDistributedLock} for detail
 * stats.
 * <li> `read_lock/*`: metrics about the locks used by readers. See {@link ZKDistributedLock} for
 * detail stats.
 * <li> `logsegments/*`: metrics about basic operations on log segments. See {@link BKLogHandler} for details.
 * <li> `segments/*`: metrics about write operations on log segments. See {@link BKLogWriteHandler} for details.
 * <li> `readahead_worker/*`: metrics about readahead workers used by readers. See {@link BKLogReadHandler}
 * for details.
 * </ul>
 */
class BKDistributedLogManager implements DistributedLogManager {
    static final Logger LOG = LoggerFactory.getLogger(BKDistributedLogManager.class);

    static final Function<LogRecordWithDLSN, Long> RECORD_2_TXID_FUNCTION =
            new Function<LogRecordWithDLSN, Long>() {
                @Override
                public Long apply(LogRecordWithDLSN record) {
                    return record.getTransactionId();
                }
            };

    static final Function<LogRecordWithDLSN, DLSN> RECORD_2_DLSN_FUNCTION =
            new Function<LogRecordWithDLSN, DLSN>() {
                @Override
                public DLSN apply(LogRecordWithDLSN record) {
                    return record.getDlsn();
                }
            };

    private final URI uri;
    private final String name;
    private final String clientId;
    private final int regionId;
    private final String streamIdentifier;
    private final DistributedLogConfiguration conf;
    private final DynamicDistributedLogConfiguration dynConf;
    private final NamespaceDriver driver;
    private Promise<Void> closePromise;
    private final OrderedScheduler scheduler;
    private final FeatureProvider featureProvider;
    private final AsyncFailureInjector failureInjector;
    private final StatsLogger statsLogger;
    private final StatsLogger perLogStatsLogger;
    final AlertStatsLogger alertStatsLogger;

    // log segment metadata cache
    private final LogSegmentMetadataCache logSegmentMetadataCache;

    //
    // Writer Related Variables
    //
    private final PermitLimiter writeLimiter;

    //
    // Reader Related Variables
    ///
    // read handler for listener.
    private BKLogReadHandler readHandlerForListener = null;
    private final PendingReaders pendingReaders;

    // resource to close
    private final Optional<AsyncCloseable> resourcesCloseable;

    /**
     * Create a {@link DistributedLogManager} with supplied resources.
     *
     * @param name log name
     * @param conf distributedlog configuration
     * @param dynConf dynamic distributedlog configuration
     * @param uri uri location for the log
     * @param driver namespace driver
     * @param logSegmentMetadataCache log segment metadata cache
     * @param scheduler ordered scheduled used by readers and writers
     * @param clientId client id that used to initiate the locks
     * @param regionId region id that would be encrypted as part of log segment metadata
     *                 to indicate which region that the log segment will be created
     * @param writeLimiter write limiter
     * @param featureProvider provider to offer features
     * @param statsLogger stats logger to receive stats
     * @param perLogStatsLogger stats logger to receive per log stats
     * @throws IOException
     */
    BKDistributedLogManager(String name,
                            DistributedLogConfiguration conf,
                            DynamicDistributedLogConfiguration dynConf,
                            URI uri,
                            NamespaceDriver driver,
                            LogSegmentMetadataCache logSegmentMetadataCache,
                            OrderedScheduler scheduler,
                            String clientId,
                            Integer regionId,
                            PermitLimiter writeLimiter,
                            FeatureProvider featureProvider,
                            AsyncFailureInjector failureInjector,
                            StatsLogger statsLogger,
                            StatsLogger perLogStatsLogger,
                            Optional<AsyncCloseable> resourcesCloseable) {
        this.name = name;
        this.conf = conf;
        this.dynConf = dynConf;
        this.uri = uri;
        this.driver = driver;
        this.logSegmentMetadataCache = logSegmentMetadataCache;
        this.scheduler = scheduler;
        this.statsLogger = statsLogger;
        this.perLogStatsLogger = BroadCastStatsLogger.masterslave(perLogStatsLogger, statsLogger);
        this.pendingReaders = new PendingReaders(scheduler);
        this.regionId = regionId;
        this.clientId = clientId;
        this.streamIdentifier = conf.getUnpartitionedStreamName();
        this.writeLimiter = writeLimiter;
        // Feature Provider
        this.featureProvider = featureProvider;
        // Failure Injector
        this.failureInjector = failureInjector;
        // Stats
        this.alertStatsLogger = new AlertStatsLogger(this.perLogStatsLogger, "dl_alert");
        this.resourcesCloseable = resourcesCloseable;
    }

    @Override
    public String getStreamName() {
        return name;
    }

    @Override
    public NamespaceDriver getNamespaceDriver() {
        return driver;
    }

    URI getUri() {
        return uri;
    }

    DistributedLogConfiguration getConf() {
        return conf;
    }

    OrderedScheduler getScheduler() {
        return scheduler;
    }

    AsyncFailureInjector getFailureInjector() {
        return failureInjector;
    }

    //
    // Test Methods
    //

    @VisibleForTesting
    LogStreamMetadataStore getWriterMetadataStore() {
        return driver.getLogStreamMetadataStore(WRITER);
    }

    @VisibleForTesting
    LogSegmentEntryStore getReaderEntryStore() {
        return driver.getLogSegmentEntryStore(READER);
    }

    @VisibleForTesting
    FeatureProvider getFeatureProvider() {
        return this.featureProvider;
    }

    private synchronized BKLogReadHandler getReadHandlerAndRegisterListener(
            boolean create, LogSegmentListener listener) {
        if (null == readHandlerForListener && create) {
            readHandlerForListener = createReadHandler();
            readHandlerForListener.registerListener(listener);
            // start fetch the log segments after created the listener
            readHandlerForListener.asyncStartFetchLogSegments();
            return readHandlerForListener;
        }
        if (null != readHandlerForListener && null != listener) {
            readHandlerForListener.registerListener(listener);
        }
        return readHandlerForListener;
    }

    @Override
    public List<LogSegmentMetadata> getLogSegments() throws IOException {
        return FutureUtils.result(getLogSegmentsAsync());
    }

    protected Future<List<LogSegmentMetadata>> getLogSegmentsAsync() {
        final BKLogReadHandler readHandler = createReadHandler();
        return readHandler.readLogSegmentsFromStore(
                LogSegmentMetadata.COMPARATOR,
                LogSegmentFilter.DEFAULT_FILTER,
                null)
                .map(GetVersionedValueFunction.GET_LOGSEGMENT_LIST_FUNC)
                .ensure(CloseAsyncCloseableFunction.of(readHandler));
    }

    @Override
    public void registerListener(LogSegmentListener listener) throws IOException {
        getReadHandlerAndRegisterListener(true, listener);
    }

    @Override
    public synchronized void unregisterListener(LogSegmentListener listener) {
        if (null != readHandlerForListener) {
            readHandlerForListener.unregisterListener(listener);
        }
    }

    public void checkClosedOrInError(String operation) throws AlreadyClosedException {
        synchronized (this) {
            if (null != closePromise) {
                throw new AlreadyClosedException("Executing " + operation + " on already closed DistributedLogManager");
            }
        }
    }

    // Create Read Handler

    synchronized BKLogReadHandler createReadHandler() {
        Optional<String> subscriberId = Optional.absent();
        return createReadHandler(subscriberId, false);
    }

    synchronized BKLogReadHandler createReadHandler(Optional<String> subscriberId) {
        return createReadHandler(subscriberId, false);
    }

    synchronized BKLogReadHandler createReadHandler(Optional<String> subscriberId,
                                                    boolean isHandleForReading) {
        return createReadHandler(
                subscriberId,
                null,
                isHandleForReading);
    }

    synchronized BKLogReadHandler createReadHandler(Optional<String> subscriberId,
                                                    AsyncNotification notification,
                                                    boolean isHandleForReading) {
        LogMetadataForReader logMetadata = LogMetadataForReader.of(uri, name, streamIdentifier);
        return new BKLogReadHandler(
                logMetadata,
                subscriberId,
                conf,
                dynConf,
                driver.getLogStreamMetadataStore(READER),
                logSegmentMetadataCache,
                driver.getLogSegmentEntryStore(READER),
                scheduler,
                alertStatsLogger,
                statsLogger,
                perLogStatsLogger,
                clientId,
                notification,
                isHandleForReading);
    }

    // Create Ledger Allocator


    // Create Write Handler

    public BKLogWriteHandler createWriteHandler(boolean lockHandler)
            throws IOException {
        return FutureUtils.result(asyncCreateWriteHandler(lockHandler));
    }

    Future<BKLogWriteHandler> asyncCreateWriteHandler(final boolean lockHandler) {
        // Fetching Log Metadata (create if not exists)
        return driver.getLogStreamMetadataStore(WRITER).getLog(
                uri,
                name,
                true,
                conf.getCreateStreamIfNotExists()
        ).flatMap(new AbstractFunction1<LogMetadataForWriter, Future<BKLogWriteHandler>>() {
            @Override
            public Future<BKLogWriteHandler> apply(LogMetadataForWriter logMetadata) {
                Promise<BKLogWriteHandler> createPromise = new Promise<BKLogWriteHandler>();
                createWriteHandler(logMetadata, lockHandler, createPromise);
                return createPromise;
            }
        });
    }

    private void createWriteHandler(LogMetadataForWriter logMetadata,
                                    boolean lockHandler,
                                    final Promise<BKLogWriteHandler> createPromise) {
        // Build the locks
        DistributedLock lock;
        if (conf.isWriteLockEnabled()) {
            lock = driver.getLogStreamMetadataStore(WRITER).createWriteLock(logMetadata);
        } else {
            lock = NopDistributedLock.INSTANCE;
        }

        Allocator<LogSegmentEntryWriter, Object> segmentAllocator;
        try {
            segmentAllocator = driver.getLogSegmentEntryStore(WRITER)
                    .newLogSegmentAllocator(logMetadata, dynConf);
        } catch (IOException ioe) {
            FutureUtils.setException(createPromise, ioe);
            return;
        }

        // Make sure writer handler created before resources are initialized
        final BKLogWriteHandler writeHandler = new BKLogWriteHandler(
                logMetadata,
                conf,
                driver.getLogStreamMetadataStore(WRITER),
                logSegmentMetadataCache,
                driver.getLogSegmentEntryStore(WRITER),
                scheduler,
                segmentAllocator,
                statsLogger,
                perLogStatsLogger,
                alertStatsLogger,
                clientId,
                regionId,
                writeLimiter,
                featureProvider,
                dynConf,
                lock);
        if (lockHandler) {
            writeHandler.lockHandler().addEventListener(new FutureEventListener<DistributedLock>() {
                @Override
                public void onSuccess(DistributedLock lock) {
                    FutureUtils.setValue(createPromise, writeHandler);
                }

                @Override
                public void onFailure(final Throwable cause) {
                    writeHandler.asyncClose().ensure(new AbstractFunction0<BoxedUnit>() {
                        @Override
                        public BoxedUnit apply() {
                            FutureUtils.setException(createPromise, cause);
                            return BoxedUnit.UNIT;
                        }
                    });
                }
            });
        } else {
            FutureUtils.setValue(createPromise, writeHandler);
        }
    }

    PermitManager getLogSegmentRollingPermitManager() {
        return driver.getLogStreamMetadataStore(WRITER).getPermitManager();
    }

    <T> Future<T> processReaderOperation(final Function<BKLogReadHandler, Future<T>> func) {
        return scheduler.apply(new ExceptionalFunction0<BKLogReadHandler>() {
            @Override
            public BKLogReadHandler applyE() throws Throwable {
                return getReadHandlerAndRegisterListener(true, null);
            }
        }).flatMap(new ExceptionalFunction<BKLogReadHandler, Future<T>>() {
            @Override
            public Future<T> applyE(final BKLogReadHandler readHandler) throws Throwable {
                return func.apply(readHandler);
            }
        });
    }

    /**
     * Check if an end of stream marker was added to the stream
     * A stream with an end of stream marker cannot be appended to
     *
     * @return true if the marker was added to the stream, false otherwise
     */
    @Override
    public boolean isEndOfStreamMarked() throws IOException {
        checkClosedOrInError("isEndOfStreamMarked");
        long lastTxId = FutureUtils.result(getLastLogRecordAsyncInternal(false, true)).getTransactionId();
        return lastTxId == DistributedLogConstants.MAX_TXID;
    }

    /**
     * Begin appending to the end of the log stream which is being treated as a sequence of bytes
     *
     * @return the writer interface to generate log records
     */
    public AppendOnlyStreamWriter getAppendOnlyStreamWriter() throws IOException {
        long position;
        try {
            position = FutureUtils.result(getLastLogRecordAsyncInternal(true, false)).getTransactionId();
            if (DistributedLogConstants.INVALID_TXID == position ||
                DistributedLogConstants.EMPTY_LOGSEGMENT_TX_ID == position) {
                position = 0;
            }
        } catch (LogEmptyException ex) {
            position = 0;
        } catch (LogNotFoundException ex) {
            position = 0;
        }
        return new AppendOnlyStreamWriter(startAsyncLogSegmentNonPartitioned(), position);
    }

    /**
     * Get a reader to read a log stream as a sequence of bytes
     *
     * @return the writer interface to generate log records
     */
    public AppendOnlyStreamReader getAppendOnlyStreamReader() throws IOException {
        return new AppendOnlyStreamReader(this);
    }

    /**
     * Begin writing to the log stream identified by the name
     *
     * @return the writer interface to generate log records
     */
    @Override
    public BKSyncLogWriter startLogSegmentNonPartitioned() throws IOException {
        checkClosedOrInError("startLogSegmentNonPartitioned");
        BKSyncLogWriter writer = new BKSyncLogWriter(conf, dynConf, this);
        boolean success = false;
        try {
            writer.createAndCacheWriteHandler();
            BKLogWriteHandler writeHandler = writer.getWriteHandler();
            FutureUtils.result(writeHandler.lockHandler());
            success = true;
            return writer;
        } finally {
            if (!success) {
                writer.abort();
            }
        }
    }

    /**
     * Begin writing to the log stream identified by the name
     *
     * @return the writer interface to generate log records
     */
    @Override
    public BKAsyncLogWriter startAsyncLogSegmentNonPartitioned() throws IOException {
        return (BKAsyncLogWriter) FutureUtils.result(openAsyncLogWriter());
    }

    @Override
    public Future<AsyncLogWriter> openAsyncLogWriter() {
        try {
            checkClosedOrInError("startLogSegmentNonPartitioned");
        } catch (AlreadyClosedException e) {
            return Future.exception(e);
        }

        Future<BKLogWriteHandler> createWriteHandleFuture;
        synchronized (this) {
            // 1. create the locked write handler
            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) {
                    // 2. create the writer with the handler
                    writer = new BKAsyncLogWriter(
                            conf,
                            dynConf,
                            BKDistributedLogManager.this,
                            writeHandler,
                            featureProvider,
                            statsLogger);
                }
                // 3. recover the incomplete log segments
                return writeHandler.recoverIncompleteLogSegments()
                        .map(new AbstractFunction1<Long, AsyncLogWriter>() {
                            @Override
                            public AsyncLogWriter apply(Long lastTxId) {
                                // 4. update last tx id if successfully recovered
                                writer.setLastTxId(lastTxId);
                                return writer;
                            }
                        }).onFailure(new AbstractFunction1<Throwable, BoxedUnit>() {
                            @Override
                            public BoxedUnit apply(Throwable cause) {
                                // 5. close the writer if recovery failed
                                writer.asyncAbort();
                                return BoxedUnit.UNIT;
                            }
                        });
            }
        });
    }

    @Override
    public Future<DLSN> getDLSNNotLessThanTxId(final long fromTxnId) {
        return getLogSegmentsAsync().flatMap(new AbstractFunction1<List<LogSegmentMetadata>, Future<DLSN>>() {
            @Override
            public Future<DLSN> apply(List<LogSegmentMetadata> segments) {
                return getDLSNNotLessThanTxId(fromTxnId, segments);
            }
        });
    }

    private Future<DLSN> getDLSNNotLessThanTxId(long fromTxnId,
                                                final List<LogSegmentMetadata> segments) {
        if (segments.isEmpty()) {
            return getLastDLSNAsync();
        }
        final int segmentIdx = DLUtils.findLogSegmentNotLessThanTxnId(segments, fromTxnId);
        if (segmentIdx < 0) {
            return Future.value(new DLSN(segments.get(0).getLogSegmentSequenceNumber(), 0L, 0L));
        }
        return getDLSNNotLessThanTxIdInSegment(
                fromTxnId,
                segmentIdx,
                segments,
                driver.getLogSegmentEntryStore(READER)
        );
    }

    private Future<DLSN> getDLSNNotLessThanTxIdInSegment(final long fromTxnId,
                                                         final int segmentIdx,
                                                         final List<LogSegmentMetadata> segments,
                                                         final LogSegmentEntryStore entryStore) {
        final LogSegmentMetadata segment = segments.get(segmentIdx);
        return ReadUtils.getLogRecordNotLessThanTxId(
                name,
                segment,
                fromTxnId,
                scheduler,
                entryStore,
                Math.max(2, dynConf.getReadAheadBatchSize())
        ).flatMap(new AbstractFunction1<Optional<LogRecordWithDLSN>, Future<DLSN>>() {
            @Override
            public Future<DLSN> apply(Optional<LogRecordWithDLSN> foundRecord) {
                if (foundRecord.isPresent()) {
                    return Future.value(foundRecord.get().getDlsn());
                }
                if ((segments.size() - 1) == segmentIdx) {
                    return getLastLogRecordAsync().map(new AbstractFunction1<LogRecordWithDLSN, DLSN>() {
                        @Override
                        public DLSN apply(LogRecordWithDLSN record) {
                            if (record.getTransactionId() >= fromTxnId) {
                                return record.getDlsn();
                            }
                            return record.getDlsn().getNextDLSN();
                        }
                    });
                } else {
                    return getDLSNNotLessThanTxIdInSegment(
                            fromTxnId,
                            segmentIdx + 1,
                            segments,
                            entryStore);
                }
            }
        });
    }

    /**
     * Get the input stream starting with fromTxnId for the specified log
     *
     * @param fromTxnId - the first transaction id we want to read
     * @return the stream starting with transaction fromTxnId
     * @throws IOException if a stream cannot be found.
     */
    @Override
    public LogReader getInputStream(long fromTxnId)
        throws IOException {
        return getInputStreamInternal(fromTxnId);
    }

    @Override
    public LogReader getInputStream(DLSN fromDLSN) throws IOException {
        return getInputStreamInternal(fromDLSN, Optional.<Long>absent());
    }

    @Override
    public AsyncLogReader getAsyncLogReader(long fromTxnId) throws IOException {
        return FutureUtils.result(openAsyncLogReader(fromTxnId));
    }

    /**
     * Opening a log reader positioning by transaction id <code>fromTxnId</code>.
     *
     * <p>
     * - retrieve log segments for the stream
     * - if the log segment list is empty, positioning by the last dlsn
     * - otherwise, find the first log segment that contains the records whose transaction ids are not less than
     *   the provided transaction id <code>fromTxnId</code>
     *   - if all log segments' records' transaction ids are more than <code>fromTxnId</code>, positioning
     *     on the first record.
     *   - otherwise, search the log segment to find the log record
     *     - if the log record is found, positioning the reader by that found record's dlsn
     *     - otherwise, positioning by the last dlsn
     * </p>
     *
     * @see DLUtils#findLogSegmentNotLessThanTxnId(List, long)
     * @see ReadUtils#getLogRecordNotLessThanTxId(String, LogSegmentMetadata, long, ExecutorService, LogSegmentEntryStore, int)
     * @param fromTxnId
     *          transaction id to start reading from
     * @return future representing the open result.
     */
    @Override
    public Future<AsyncLogReader> openAsyncLogReader(long fromTxnId) {
        final Promise<DLSN> dlsnPromise = new Promise<DLSN>();
        getDLSNNotLessThanTxId(fromTxnId).addEventListener(new FutureEventListener<DLSN>() {

            @Override
            public void onSuccess(DLSN dlsn) {
                dlsnPromise.setValue(dlsn);
            }

            @Override
            public void onFailure(Throwable cause) {
                if (cause instanceof LogEmptyException) {
                    dlsnPromise.setValue(DLSN.InitialDLSN);
                } else {
                    dlsnPromise.setException(cause);
                }
            }
        });
        return dlsnPromise.flatMap(new AbstractFunction1<DLSN, Future<AsyncLogReader>>() {
            @Override
            public Future<AsyncLogReader> apply(DLSN dlsn) {
                return openAsyncLogReader(dlsn);
            }
        });
    }

    @Override
    public AsyncLogReader getAsyncLogReader(DLSN fromDLSN) throws IOException {
        return FutureUtils.result(openAsyncLogReader(fromDLSN));
    }

    @Override
    public Future<AsyncLogReader> openAsyncLogReader(DLSN fromDLSN) {
        Optional<String> subscriberId = Optional.absent();
        AsyncLogReader reader = new BKAsyncLogReader(
                this,
                scheduler,
                fromDLSN,
                subscriberId,
                false,
                statsLogger);
        pendingReaders.add(reader);
        return Future.value(reader);
    }

    /**
     * Note the lock here is a sort of elective exclusive lock. I.e. acquiring this lock will only prevent other
     * people who try to acquire the lock from reading from the stream. Normal readers (and writers) will not be
     * blocked.
     */
    @Override
    public Future<AsyncLogReader> getAsyncLogReaderWithLock(final DLSN fromDLSN) {
        Optional<String> subscriberId = Optional.absent();
        return getAsyncLogReaderWithLock(Optional.of(fromDLSN), subscriberId);
    }

    @Override
    public Future<AsyncLogReader> getAsyncLogReaderWithLock(final DLSN fromDLSN, final String subscriberId) {
        return getAsyncLogReaderWithLock(Optional.of(fromDLSN), Optional.of(subscriberId));
    }

    @Override
    public Future<AsyncLogReader> getAsyncLogReaderWithLock(String subscriberId) {
        Optional<DLSN> fromDLSN = Optional.absent();
        return getAsyncLogReaderWithLock(fromDLSN, Optional.of(subscriberId));
    }

    protected Future<AsyncLogReader> getAsyncLogReaderWithLock(final Optional<DLSN> fromDLSN,
                                                               final Optional<String> subscriberId) {
        if (!fromDLSN.isPresent() && !subscriberId.isPresent()) {
            return Future.exception(new UnexpectedException("Neither from dlsn nor subscriber id is provided."));
        }
        final BKAsyncLogReader reader = new BKAsyncLogReader(
                BKDistributedLogManager.this,
                scheduler,
                fromDLSN.isPresent() ? fromDLSN.get() : DLSN.InitialDLSN,
                subscriberId,
                false,
                statsLogger);
        pendingReaders.add(reader);
        final Future<Void> lockFuture = reader.lockStream();
        final Promise<AsyncLogReader> createPromise = new Promise<AsyncLogReader>(
                new Function<Throwable, BoxedUnit>() {
            @Override
            public BoxedUnit apply(Throwable cause) {
                // cancel the lock when the creation future is cancelled
                lockFuture.cancel();
                return BoxedUnit.UNIT;
            }
        });
        // lock the stream - fetch the last commit position on success
        lockFuture.flatMap(new Function<Void, Future<AsyncLogReader>>() {
            @Override
            public Future<AsyncLogReader> apply(Void complete) {
                if (fromDLSN.isPresent()) {
                    return Future.value((AsyncLogReader) reader);
                }
                LOG.info("Reader {} @ {} reading last commit position from subscription store after acquired lock.",
                        subscriberId.get(), name);
                // we acquired lock
                final SubscriptionsStore subscriptionsStore = driver.getSubscriptionsStore(getStreamName());
                return subscriptionsStore.getLastCommitPosition(subscriberId.get())
                        .map(new ExceptionalFunction<DLSN, AsyncLogReader>() {
                    @Override
                    public AsyncLogReader applyE(DLSN lastCommitPosition) throws UnexpectedException {
                        LOG.info("Reader {} @ {} positioned to last commit position {}.",
                                new Object[] { subscriberId.get(), name, lastCommitPosition });
                        reader.setStartDLSN(lastCommitPosition);
                        return reader;
                    }
                });
            }
        }).addEventListener(new FutureEventListener<AsyncLogReader>() {
            @Override
            public void onSuccess(AsyncLogReader r) {
                pendingReaders.remove(reader);
                FutureUtils.setValue(createPromise, r);
            }

            @Override
            public void onFailure(final Throwable cause) {
                pendingReaders.remove(reader);
                reader.asyncClose().ensure(new AbstractFunction0<BoxedUnit>() {
                    @Override
                    public BoxedUnit apply() {
                        FutureUtils.setException(createPromise, cause);
                        return BoxedUnit.UNIT;
                    }
                });
            }
        });
        return createPromise;
    }

    /**
     * Get the input stream starting with fromTxnId for the specified log
     *
     * @param fromTxnId
     *          transaction id to start reading from
     * @return log reader
     * @throws IOException
     */
    LogReader getInputStreamInternal(long fromTxnId)
        throws IOException {
        DLSN fromDLSN;
        try {
            fromDLSN = FutureUtils.result(getDLSNNotLessThanTxId(fromTxnId));
        } catch (LogEmptyException lee) {
            fromDLSN = DLSN.InitialDLSN;
        }
        return getInputStreamInternal(fromDLSN, Optional.of(fromTxnId));
    }

    LogReader getInputStreamInternal(DLSN fromDLSN, Optional<Long> fromTxnId)
            throws IOException {
        LOG.info("Create sync reader starting from {}", fromDLSN);
        checkClosedOrInError("getInputStream");
        return new BKSyncLogReader(
                conf,
                this,
                fromDLSN,
                fromTxnId,
                statsLogger);
    }

    /**
     * Get the last log record in the stream
     *
     * @return the last log record in the stream
     * @throws java.io.IOException if a stream cannot be found.
     */
    @Override
    public LogRecordWithDLSN getLastLogRecord() throws IOException {
        checkClosedOrInError("getLastLogRecord");
        return FutureUtils.result(getLastLogRecordAsync());
    }

    @Override
    public long getFirstTxId() throws IOException {
        checkClosedOrInError("getFirstTxId");
        return FutureUtils.result(getFirstRecordAsyncInternal()).getTransactionId();
    }

    @Override
    public long getLastTxId() throws IOException {
        checkClosedOrInError("getLastTxId");
        return FutureUtils.result(getLastTxIdAsync());
    }

    @Override
    public DLSN getLastDLSN() throws IOException {
        checkClosedOrInError("getLastDLSN");
        return FutureUtils.result(getLastLogRecordAsyncInternal(false, false)).getDlsn();
    }

    /**
     * Get Latest log record in the log
     *
     * @return latest log record
     */
    @Override
    public Future<LogRecordWithDLSN> getLastLogRecordAsync() {
        return getLastLogRecordAsyncInternal(false, false);
    }

    private Future<LogRecordWithDLSN> getLastLogRecordAsyncInternal(final boolean recover,
                                                                    final boolean includeEndOfStream) {
        return processReaderOperation(new Function<BKLogReadHandler, Future<LogRecordWithDLSN>>() {
            @Override
            public Future<LogRecordWithDLSN> apply(final BKLogReadHandler ledgerHandler) {
                return ledgerHandler.getLastLogRecordAsync(recover, includeEndOfStream);
            }
        });
    }

    /**
     * Get Latest Transaction Id in the log
     *
     * @return latest transaction id
     */
    @Override
    public Future<Long> getLastTxIdAsync() {
        return getLastLogRecordAsyncInternal(false, false)
                .map(RECORD_2_TXID_FUNCTION);
    }

    /**
     * Get first DLSN in the log.
     *
     * @return first dlsn in the stream
     */
    @Override
    public Future<DLSN> getFirstDLSNAsync() {
        return getFirstRecordAsyncInternal().map(RECORD_2_DLSN_FUNCTION);
    }

    private Future<LogRecordWithDLSN> getFirstRecordAsyncInternal() {
        return processReaderOperation(new Function<BKLogReadHandler, Future<LogRecordWithDLSN>>() {
            @Override
            public Future<LogRecordWithDLSN> apply(final BKLogReadHandler ledgerHandler) {
                return ledgerHandler.asyncGetFirstLogRecord();
            }
        });
    }

    /**
     * Get Latest DLSN in the log.
     *
     * @return latest transaction id
     */
    @Override
    public Future<DLSN> getLastDLSNAsync() {
        return getLastLogRecordAsyncInternal(false, false)
                .map(RECORD_2_DLSN_FUNCTION);
    }

    /**
     * Get the number of log records in the active portion of the log
     * Any log segments that have already been truncated will not be included
     *
     * @return number of log records
     * @throws IOException
     */
    @Override
    public long getLogRecordCount() throws IOException {
        checkClosedOrInError("getLogRecordCount");
        return FutureUtils.result(getLogRecordCountAsync(DLSN.InitialDLSN));
    }

    /**
     * Get the number of log records in the active portion of the log asynchronously.
     * Any log segments that have already been truncated will not be included
     *
     * @return future number of log records
     * @throws IOException
     */
    @Override
    public Future<Long> getLogRecordCountAsync(final DLSN beginDLSN) {
        return processReaderOperation(new Function<BKLogReadHandler, Future<Long>>() {
                    @Override
                    public Future<Long> apply(BKLogReadHandler ledgerHandler) {
                        return ledgerHandler.asyncGetLogRecordCount(beginDLSN);
                    }
                });
    }

    @Override
    public void recover() throws IOException {
        recoverInternal(conf.getUnpartitionedStreamName());
    }

    /**
     * Recover a specified stream within the log container
     * The writer implicitly recovers a topic when it resumes writing.
     * This allows applications to recover a container explicitly so
     * that application may read a fully recovered log before resuming
     * the writes
     *
     * @throws IOException if the recovery fails
     */
    private void recoverInternal(String streamIdentifier) throws IOException {
        checkClosedOrInError("recoverInternal");
        BKLogWriteHandler ledgerHandler = createWriteHandler(true);
        try {
            FutureUtils.result(ledgerHandler.recoverIncompleteLogSegments());
        } finally {
            Utils.closeQuietly(ledgerHandler);
        }
    }

    /**
     * Delete all the partitions of the specified log
     *
     * @throws IOException if the deletion fails
     */
    @Override
    public void delete() throws IOException {
        FutureUtils.result(driver.getLogStreamMetadataStore(WRITER)
                .deleteLog(uri, getStreamName()));
    }

    /**
     * The DistributedLogManager may archive/purge any logs for transactionId
     * less than or equal to minImageTxId.
     * This is to be used only when the client explicitly manages deletion. If
     * the cleanup policy is based on sliding time window, then this method need
     * not be called.
     *
     * @param minTxIdToKeep the earliest txid that must be retained
     * @throws IOException if purging fails
     */
    @Override
    public void purgeLogsOlderThan(long minTxIdToKeep) throws IOException {
        Preconditions.checkArgument(minTxIdToKeep > 0, "Invalid transaction id " + minTxIdToKeep);
        checkClosedOrInError("purgeLogSegmentsOlderThan");
        BKLogWriteHandler ledgerHandler = createWriteHandler(true);
        try {
            LOG.info("Purging logs for {} older than {}", ledgerHandler.getFullyQualifiedName(), minTxIdToKeep);
            FutureUtils.result(ledgerHandler.purgeLogSegmentsOlderThanTxnId(minTxIdToKeep));
        } finally {
            Utils.closeQuietly(ledgerHandler);
        }
    }

    static class PendingReaders implements AsyncCloseable {

        final ExecutorService executorService;
        final Set<AsyncCloseable> readers = new HashSet<AsyncCloseable>();

        PendingReaders(ExecutorService executorService) {
            this.executorService = executorService;
        }

        public synchronized void remove(AsyncCloseable reader) {
            readers.remove(reader);
        }

        public synchronized void add(AsyncCloseable reader) {
            readers.add(reader);
        }

        @Override
        public Future<Void> asyncClose() {
            return Utils.closeSequence(executorService, true, readers.toArray(new AsyncLogReader[readers.size()]))
                    .onSuccess(new AbstractFunction1<Void, BoxedUnit>() {
                        @Override
                        public BoxedUnit apply(Void value) {
                            readers.clear();
                            return BoxedUnit.UNIT;
                        }
                    });
        }
    };

    /**
     * Close the distributed log manager, freeing any resources it may hold.
     */
    @Override
    public Future<Void> asyncClose() {
        Promise<Void> closeFuture;
        BKLogReadHandler readHandlerToClose;
        synchronized (this) {
            if (null != closePromise) {
                return closePromise;
            }
            closeFuture = closePromise = new Promise<Void>();
            readHandlerToClose = readHandlerForListener;
        }

        Future<Void> closeResult = Utils.closeSequence(null, true,
                readHandlerToClose,
                pendingReaders,
                resourcesCloseable.or(AsyncCloseable.NULL));
        closeResult.proxyTo(closeFuture);
        return closeFuture;
    }

    @Override
    public void close() throws IOException {
        FutureUtils.result(asyncClose());
    }

    @Override
    public String toString() {
        return String.format("DLM:%s:%s", getUri(), getStreamName());
    }

    public void raiseAlert(String msg, Object... args) {
        alertStatsLogger.raise(msg, args);
    }

    @Override
    public SubscriptionsStore getSubscriptionsStore() {
        return driver.getSubscriptionsStore(getStreamName());
    }

}