/**
 * 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 com.twitter.distributedlog;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.twitter.distributedlog.bk.DynamicQuorumConfigProvider;
import com.twitter.distributedlog.bk.LedgerAllocator;
import com.twitter.distributedlog.bk.LedgerAllocatorDelegator;
import com.twitter.distributedlog.bk.QuorumConfigProvider;
import com.twitter.distributedlog.bk.SimpleLedgerAllocator;
import com.twitter.distributedlog.callback.LogSegmentListener;
import com.twitter.distributedlog.config.DynamicDistributedLogConfiguration;
import com.twitter.distributedlog.exceptions.AlreadyClosedException;
import com.twitter.distributedlog.exceptions.DLInterruptedException;
import com.twitter.distributedlog.exceptions.LogEmptyException;
import com.twitter.distributedlog.exceptions.LogNotFoundException;
import com.twitter.distributedlog.exceptions.UnexpectedException;
import com.twitter.distributedlog.impl.ZKLogSegmentMetadataStore;
import com.twitter.distributedlog.impl.metadata.ZKLogMetadataForReader;
import com.twitter.distributedlog.impl.metadata.ZKLogMetadataForWriter;
import com.twitter.distributedlog.io.AsyncCloseable;
import com.twitter.distributedlog.lock.DistributedLock;
import com.twitter.distributedlog.lock.NopDistributedLock;
import com.twitter.distributedlog.lock.SessionLockFactory;
import com.twitter.distributedlog.lock.ZKDistributedLock;
import com.twitter.distributedlog.lock.ZKSessionLockFactory;
import com.twitter.distributedlog.logsegment.LogSegmentMetadataStore;
import com.twitter.distributedlog.metadata.BKDLConfig;
import com.twitter.distributedlog.stats.BroadCastStatsLogger;
import com.twitter.distributedlog.stats.ReadAheadExceptionsLogger;
import com.twitter.distributedlog.subscription.SubscriptionStateStore;
import com.twitter.distributedlog.subscription.SubscriptionsStore;
import com.twitter.distributedlog.subscription.ZKSubscriptionStateStore;
import com.twitter.distributedlog.subscription.ZKSubscriptionsStore;
import com.twitter.distributedlog.util.ConfUtils;
import com.twitter.distributedlog.util.DLUtils;
import com.twitter.distributedlog.util.FutureUtils;
import com.twitter.distributedlog.util.MonitoredFuturePool;
import com.twitter.distributedlog.util.OrderedScheduler;
import com.twitter.distributedlog.util.PermitLimiter;
import com.twitter.distributedlog.util.PermitManager;
import com.twitter.distributedlog.util.SchedulerUtils;
import com.twitter.distributedlog.util.Utils;
import com.twitter.util.ExceptionalFunction;
import com.twitter.util.ExceptionalFunction0;
import com.twitter.util.ExecutorServiceFuturePool;
import com.twitter.util.Function;
import com.twitter.util.Future;
import com.twitter.util.FuturePool;
import com.twitter.util.FutureEventListener;
import com.twitter.util.Promise;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.ZooKeeper;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
import org.jboss.netty.util.HashedWheelTimer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.runtime.AbstractFunction0;
import scala.runtime.AbstractFunction1;
import scala.runtime.BoxedUnit;

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.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * <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 BKAsyncLogReaderDLSN} 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 extends ZKMetadataAccessor implements DistributedLogManager {
    static final Logger LOG = LoggerFactory.getLogger(BKDistributedLogManager.class);

    static void createLog(DistributedLogConfiguration conf, ZooKeeperClient zkc, URI uri, String streamName)
            throws IOException, InterruptedException {
        Future<ZKLogMetadataForWriter> createFuture = ZKLogMetadataForWriter.of(
                        uri, streamName, conf.getUnpartitionedStreamName(), zkc.get(), zkc.getDefaultACL(), true, true);
        FutureUtils.result(createFuture);
    }

    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 String clientId;
    private final int regionId;
    private final String streamIdentifier;
    private final DistributedLogConfiguration conf;
    private final DynamicDistributedLogConfiguration dynConf;
    private Promise<Void> closePromise;
    private final OrderedScheduler scheduler;
    private final OrderedScheduler readAheadScheduler;
    private boolean ownExecutor;
    private final FeatureProvider featureProvider;
    private final StatsLogger statsLogger;
    private final StatsLogger perLogStatsLogger;
    private final AlertStatsLogger alertStatsLogger;

    // lock factory
    private SessionLockFactory lockFactory = null;

    // log segment metadata stores
    private final LogSegmentMetadataStore writerMetadataStore;
    private final LogSegmentMetadataStore readerMetadataStore;

    // bookkeeper clients
    // NOTE: The actual bookkeeper client is initialized lazily when it is referenced by
    //       {@link com.twitter.distributedlog.BookKeeperClient#get()}. So it is safe to
    //       keep builders and their client wrappers here, as they will be used when
    //       instantiating readers or writers.
    private final BookKeeperClientBuilder writerBKCBuilder;
    private final BookKeeperClient writerBKC;
    private final boolean ownWriterBKC;
    private final BookKeeperClientBuilder readerBKCBuilder;
    private final BookKeeperClient readerBKC;
    private final boolean ownReaderBKC;

    //
    // Writer Related Variables
    //
    private final LedgerAllocator ledgerAllocator;
    private final PermitLimiter writeLimiter;
    // Log Segment Rolling Manager to control rolling speed
    private final PermitManager logSegmentRollingPermitManager;
    private OrderedScheduler lockStateExecutor = null;

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

    /**
     * Create a DLM for testing.
     *
     * @param name log name
     * @param conf distributedlog configuration
     * @param uri uri location for the log
     * @param writerZKCBuilder zookeeper builder for writers
     * @param readerZKCBuilder zookeeper builder for readers
     * @param zkcForWriterBKC zookeeper builder for bookkeeper shared by writers
     * @param zkcForReaderBKC zookeeper builder for bookkeeper shared by readers
     * @param writerBKCBuilder bookkeeper builder for writers
     * @param readerBKCBuilder bookkeeper builder for readers
     * @param featureProvider provider to offer features
     * @param writeLimiter write limiter
     * @param statsLogger stats logger to receive stats
     * @throws IOException
     */
    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) throws IOException {
        this(name,
             conf,
             ConfUtils.getConstDynConf(conf),
             uri,
             writerZKCBuilder,
             readerZKCBuilder,
             zkcForWriterBKC,
             zkcForReaderBKC,
             writerBKCBuilder,
             readerBKCBuilder,
             null,
             null,
             null,
             OrderedScheduler.newBuilder().name("BKDL-" + name).corePoolSize(1).build(),
             null,
             null,
             null,
             null,
             new ReadAheadExceptionsLogger(statsLogger),
             DistributedLogConstants.UNKNOWN_CLIENT_ID,
             DistributedLogConstants.LOCAL_REGION_ID,
             null,
             writeLimiter,
             PermitManager.UNLIMITED_PERMIT_MANAGER,
             featureProvider,
             statsLogger,
             NullStatsLogger.INSTANCE);
        this.ownExecutor = true;
    }

    /**
     * Create a {@link DistributedLogManager} with supplied resources.
     *
     * @param name log name
     * @param conf distributedlog configuration
     * @param uri uri location for the log
     * @param writerZKCBuilder zookeeper builder for writers
     * @param readerZKCBuilder zookeeper builder for readers
     * @param zkcForWriterBKC zookeeper builder for bookkeeper shared by writers
     * @param zkcForReaderBKC zookeeper builder for bookkeeper shared by readers
     * @param writerBKCBuilder bookkeeper builder for writers
     * @param readerBKCBuilder bookkeeper builder for readers
     * @param lockFactory distributed lock factory
     * @param writerMetadataStore writer metadata store
     * @param readerMetadataStore reader metadata store
     * @param scheduler ordered scheduled used by readers and writers
     * @param readAheadScheduler readAhead scheduler used by readers
     * @param lockStateExecutor ordered scheduled used by locks to execute lock actions
     * @param channelFactory client socket channel factory to build bookkeeper clients
     * @param requestTimer request timer to build bookkeeper clients
     * @param readAheadExceptionsLogger stats logger to record readahead exceptions
     * @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 ledgerAllocator ledger allocator to allocate ledgers
     * @param featureProvider provider to offer features
     * @param writeLimiter write limiter
     * @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,
                            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) throws IOException {
        super(name, conf, uri, writerZKCBuilder, readerZKCBuilder, statsLogger);
        Preconditions.checkNotNull(readAheadExceptionsLogger, "No ReadAhead Stats Logger Provided.");
        this.conf = conf;
        this.dynConf = dynConf;
        this.scheduler = scheduler;
        this.lockFactory = lockFactory;
        this.lockStateExecutor = lockStateExecutor;
        this.readAheadScheduler = null == readAheadScheduler ? scheduler : readAheadScheduler;
        this.statsLogger = statsLogger;
        this.perLogStatsLogger = BroadCastStatsLogger.masterslave(perLogStatsLogger, statsLogger);
        this.ownExecutor = false;
        this.pendingReaders = new PendingReaders(scheduler);
        this.regionId = regionId;
        this.clientId = clientId;
        this.streamIdentifier = conf.getUnpartitionedStreamName();
        this.ledgerAllocator = ledgerAllocator;
        this.writeLimiter = writeLimiter;
        this.logSegmentRollingPermitManager = logSegmentRollingPermitManager;

        if (null == writerMetadataStore) {
            this.writerMetadataStore = new ZKLogSegmentMetadataStore(conf, writerZKC, scheduler);
        } else {
            this.writerMetadataStore = writerMetadataStore;
        }
        if (null == readerMetadataStore) {
            this.readerMetadataStore = new ZKLogSegmentMetadataStore(conf, readerZKC, scheduler);
        } else {
            this.readerMetadataStore = readerMetadataStore;
        }

        // create the bkc for writers
        if (null == writerBKCBuilder) {
            // resolve uri
            BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(writerZKC, uri);
            BKDLConfig.propagateConfiguration(bkdlConfig, conf);
            this.writerBKCBuilder = BookKeeperClientBuilder.newBuilder()
                    .dlConfig(conf)
                    .name(String.format("bk:%s:dlm_writer_shared", name))
                    .ledgersPath(bkdlConfig.getBkLedgersPath())
                    .channelFactory(channelFactory)
                    .requestTimer(requestTimer)
                    .statsLogger(statsLogger);
            if (null == zkcForWriterBKC) {
                this.writerBKCBuilder.zkServers(bkdlConfig.getBkZkServersForWriter());
            } else {
                this.writerBKCBuilder.zkc(zkcForWriterBKC);
            }
            this.ownWriterBKC = true;
        } else {
            this.writerBKCBuilder = writerBKCBuilder;
            this.ownWriterBKC = false;
        }
        this.writerBKC = this.writerBKCBuilder.build();

        // create the bkc for readers
        if (null == readerBKCBuilder) {
            // resolve uri
            BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(writerZKC, uri);
            BKDLConfig.propagateConfiguration(bkdlConfig, conf);
            if (bkdlConfig.getBkZkServersForWriter().equals(bkdlConfig.getBkZkServersForReader())) {
                this.readerBKCBuilder = this.writerBKCBuilder;
                this.ownReaderBKC = false;
            } else {
                this.readerBKCBuilder = BookKeeperClientBuilder.newBuilder()
                        .dlConfig(conf)
                        .name(String.format("bk:%s:dlm_reader_shared", name))
                        .ledgersPath(bkdlConfig.getBkLedgersPath())
                        .channelFactory(channelFactory)
                        .requestTimer(requestTimer)
                        .statsLogger(statsLogger);
                if (null == zkcForReaderBKC) {
                    this.readerBKCBuilder.zkServers(bkdlConfig.getBkZkServersForReader());
                } else {
                    this.readerBKCBuilder.zkc(zkcForReaderBKC);
                }
                this.ownReaderBKC = true;
            }
        } else {
            this.readerBKCBuilder = readerBKCBuilder;
            this.ownReaderBKC = false;
        }
        this.readerBKC = this.readerBKCBuilder.build();

        // Feature Provider
        this.featureProvider = featureProvider;

        // Stats
        this.alertStatsLogger = new AlertStatsLogger(this.perLogStatsLogger, "dl_alert");
        this.readAheadExceptionsLogger = readAheadExceptionsLogger;
    }

    private synchronized OrderedScheduler getLockStateExecutor(boolean createIfNull) {
        if (createIfNull && null == lockStateExecutor && ownExecutor) {
            lockStateExecutor = OrderedScheduler.newBuilder()
                    .corePoolSize(1).name("BKDL-LockState").build();
        }
        return lockStateExecutor;
    }

    private synchronized SessionLockFactory getLockFactory(boolean createIfNull) {
        if (createIfNull && null == lockFactory) {
            lockFactory = new ZKSessionLockFactory(
                    writerZKC,
                    clientId,
                    getLockStateExecutor(createIfNull),
                    conf.getZKNumRetries(),
                    conf.getLockTimeoutMilliSeconds(),
                    conf.getZKRetryBackoffStartMillis(),
                    statsLogger);
        }
        return lockFactory;
    }

    DistributedLogConfiguration getConf() {
        return conf;
    }

    OrderedScheduler getScheduler() {
        return scheduler;
    }

    @VisibleForTesting
    BookKeeperClient getWriterBKC() {
        return this.writerBKC;
    }

    @VisibleForTesting
    BookKeeperClient getReaderBKC() {
        return this.readerBKC;
    }

    @VisibleForTesting
    FuturePool getReaderFuturePool() {
        return this.readerFuturePool;
    }

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

    private synchronized BKLogReadHandler getReadHandlerForListener(boolean create) {
        if (null == readHandlerForListener && create) {
            readHandlerForListener = createReadHandler();
            readHandlerForListener.scheduleGetLedgersTask(true, true);
        }
        return readHandlerForListener;
    }

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

    protected Future<List<LogSegmentMetadata>> getLogSegmentsAsync() {
        final BKLogReadHandler readHandler = createReadHandler();
        return readHandler.asyncGetFullLedgerList(true, false).ensure(new AbstractFunction0<BoxedUnit>() {
            @Override
            public BoxedUnit apply() {
                readHandler.asyncClose();
                return BoxedUnit.UNIT;
            }
        });
    }

    @Override
    public void registerListener(LogSegmentListener listener) throws IOException {
        BKLogReadHandler readHandler = getReadHandlerForListener(true);
        readHandler.registerListener(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");
            }
        }
        if (null != writerBKC) {
            writerBKC.checkClosedOrInError();
        }
        if (null != readerBKC) {
            readerBKC.checkClosedOrInError();
        }
    }

    // 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,
                getLockStateExecutor(true),
                null,
                true, /* deserialize record set */
                isHandleForReading);
    }

    synchronized BKLogReadHandler createReadHandler(Optional<String> subscriberId,
                                                    OrderedScheduler lockExecutor,
                                                    AsyncNotification notification,
                                                    boolean deserializeRecordSet,
                                                    boolean isHandleForReading) {
        ZKLogMetadataForReader logMetadata = ZKLogMetadataForReader.of(uri, name, streamIdentifier);
        return new BKLogReadHandler(
                logMetadata,
                subscriberId,
                conf,
                dynConf,
                readerZKCBuilder,
                readerBKCBuilder,
                readerMetadataStore,
                scheduler,
                lockExecutor,
                readAheadScheduler,
                alertStatsLogger,
                readAheadExceptionsLogger,
                statsLogger,
                perLogStatsLogger,
                clientId,
                notification,
                isHandleForReading,
                deserializeRecordSet);
    }

    // Create Ledger Allocator

    LedgerAllocator createLedgerAllocator(ZKLogMetadataForWriter logMetadata) throws IOException {
        LedgerAllocator ledgerAllocatorDelegator;
        if (!dynConf.getEnableLedgerAllocatorPool()) {
            QuorumConfigProvider quorumConfigProvider =
                    new DynamicQuorumConfigProvider(dynConf);
            LedgerAllocator allocator = new SimpleLedgerAllocator(
                    logMetadata.getAllocationPath(),
                    logMetadata.getAllocationData(),
                    quorumConfigProvider,
                    writerZKC,
                    writerBKC);
            ledgerAllocatorDelegator = new LedgerAllocatorDelegator(allocator, true);
        } else {
            ledgerAllocatorDelegator = ledgerAllocator;
        }
        return ledgerAllocatorDelegator;
    }

    // Create Write Handler

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

    Future<BKLogWriteHandler> asyncCreateWriteHandler(final boolean lockHandler) {
        final ZooKeeper zk;
        try {
            zk = writerZKC.get();
        } catch (InterruptedException e) {
            LOG.error("Failed to initialize zookeeper client : ", e);
            return Future.exception(new DLInterruptedException("Failed to initialize zookeeper client", e));
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            return Future.exception(FutureUtils.zkException(e, uri.getPath()));
        }

        boolean ownAllocator = null == ledgerAllocator;

        // Fetching Log Metadata
        Future<ZKLogMetadataForWriter> metadataFuture =
                ZKLogMetadataForWriter.of(uri, name, streamIdentifier,
                        zk, writerZKC.getDefaultACL(),
                        ownAllocator, conf.getCreateStreamIfNotExists() || ownAllocator);
        return metadataFuture.flatMap(new AbstractFunction1<ZKLogMetadataForWriter, Future<BKLogWriteHandler>>() {
            @Override
            public Future<BKLogWriteHandler> apply(ZKLogMetadataForWriter logMetadata) {
                Promise<BKLogWriteHandler> createPromise = new Promise<BKLogWriteHandler>();
                createWriteHandler(logMetadata, lockHandler, createPromise);
                return createPromise;
            }
        });
    }

    private void createWriteHandler(ZKLogMetadataForWriter logMetadata,
                                    boolean lockHandler,
                                    final Promise<BKLogWriteHandler> createPromise) {
        OrderedScheduler lockStateExecutor = getLockStateExecutor(true);
        // Build the locks
        DistributedLock lock;
        if (conf.isWriteLockEnabled()) {
            lock = new ZKDistributedLock(
                    lockStateExecutor,
                    getLockFactory(true),
                    logMetadata.getLockPath(),
                    conf.getLockTimeoutMilliSeconds(),
                    statsLogger);
        } else {
            lock = NopDistributedLock.INSTANCE;
        }
        // Build the ledger allocator
        LedgerAllocator allocator;
        try {
            allocator = createLedgerAllocator(logMetadata);
        } catch (IOException e) {
            FutureUtils.setException(createPromise, e);
            return;
        }

        // Make sure writer handler created before resources are initialized
        final BKLogWriteHandler writeHandler = new BKLogWriteHandler(
                logMetadata,
                conf,
                writerZKCBuilder,
                writerBKCBuilder,
                writerMetadataStore,
                scheduler,
                allocator,
                statsLogger,
                perLogStatsLogger,
                alertStatsLogger,
                clientId,
                regionId,
                writeLimiter,
                featureProvider,
                dynConf,
                lock);
        PermitManager manager = getLogSegmentRollingPermitManager();
        if (manager instanceof Watcher) {
            writeHandler.register((Watcher) manager);
        }
        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 logSegmentRollingPermitManager;
    }

    <T> Future<T> processReaderOperation(final Function<BKLogReadHandler, Future<T>> func) {
        initializeFuturePool(false);
        return readerFuturePool.apply(new ExceptionalFunction0<BKLogReadHandler>() {
            @Override
            public BKLogReadHandler applyE() throws Throwable {
                return getReadHandlerForListener(true);
            }
        }).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
     *
