/**
 * 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.Stopwatch;
import com.google.common.collect.Lists;
import com.twitter.distributedlog.bk.LedgerAllocator;
import com.twitter.distributedlog.config.DynamicDistributedLogConfiguration;
import com.twitter.distributedlog.exceptions.DLIllegalStateException;
import com.twitter.distributedlog.exceptions.DLInterruptedException;
import com.twitter.distributedlog.exceptions.EndOfStreamException;
import com.twitter.distributedlog.exceptions.LockingException;
import com.twitter.distributedlog.exceptions.TransactionIdOutOfOrderException;
import com.twitter.distributedlog.exceptions.UnexpectedException;
import com.twitter.distributedlog.exceptions.ZKException;
import com.twitter.distributedlog.function.GetLastTxIdFunction;
import com.twitter.distributedlog.impl.BKLogSegmentEntryWriter;
import com.twitter.distributedlog.impl.metadata.ZKLogMetadataForWriter;
import com.twitter.distributedlog.lock.DistributedLock;
import com.twitter.distributedlog.logsegment.LogSegmentMetadataStore;
import com.twitter.distributedlog.logsegment.RollingPolicy;
import com.twitter.distributedlog.logsegment.SizeBasedRollingPolicy;
import com.twitter.distributedlog.logsegment.TimeBasedRollingPolicy;
import com.twitter.distributedlog.metadata.MetadataUpdater;
import com.twitter.distributedlog.metadata.LogSegmentMetadataStoreUpdater;
import com.twitter.distributedlog.util.DLUtils;
import com.twitter.distributedlog.util.FailpointUtils;
import com.twitter.distributedlog.util.FutureUtils;
import com.twitter.distributedlog.util.FutureUtils.FutureEventListenerRunnable;
import com.twitter.distributedlog.util.OrderedScheduler;
import com.twitter.distributedlog.util.Transaction;
import com.twitter.distributedlog.util.PermitLimiter;
import com.twitter.distributedlog.util.Utils;
import com.twitter.distributedlog.zk.ZKOp;
import com.twitter.distributedlog.zk.ZKTransaction;
import com.twitter.distributedlog.zk.ZKVersionedSetOp;
import com.twitter.util.Function;
import com.twitter.util.Future;
import com.twitter.util.FutureEventListener;
import com.twitter.util.Promise;
import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.meta.ZkVersion;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.versioning.Version;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
import org.apache.zookeeper.OpResult;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.runtime.AbstractFunction1;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Charsets.UTF_8;
import static com.twitter.distributedlog.impl.ZKLogSegmentFilters.WRITE_HANDLE_FILTER;

/**
 * Log Handler for Writers.
 *
 * <h3>Metrics</h3>
 * All the metrics about log write handler are exposed under scope `segments`.
 * <ul>
 * <li> `segments`/open : opstats. latency characteristics on starting a new log segment.
 * <li> `segments`/close : opstats. latency characteristics on completing an inprogress log segment.
 * <li> `segments`/recover : opstats. latency characteristics on recovering a log segment.
 * <li> `segments`/delete : opstats. latency characteristics on deleting a log segment.
 * </ul>
 */
class BKLogWriteHandler extends BKLogHandler {
    static final Logger LOG = LoggerFactory.getLogger(BKLogReadHandler.class);

    protected final DistributedLock lock;
    protected final int ensembleSize;
    protected final int writeQuorumSize;
    protected final int ackQuorumSize;
    protected final LedgerAllocator ledgerAllocator;
    protected final MaxTxId maxTxId;
    protected final MaxLogSegmentSequenceNo maxLogSegmentSequenceNo;
    protected final boolean sanityCheckTxnId;
    protected final boolean validateLogSegmentSequenceNumber;
    protected final int regionId;
    protected final RollingPolicy rollingPolicy;
    protected Future<? extends DistributedLock> lockFuture = null;
    protected final PermitLimiter writeLimiter;
    protected final FeatureProvider featureProvider;
    protected final DynamicDistributedLogConfiguration dynConf;
    protected final MetadataUpdater metadataUpdater;
    // tracking the inprogress log segments
    protected final LinkedList<Long> inprogressLSSNs;

    // Recover Functions
    private final RecoverLogSegmentFunction recoverLogSegmentFunction =
            new RecoverLogSegmentFunction();
    private final AbstractFunction1<List<LogSegmentMetadata>, Future<Long>> recoverLogSegmentsFunction =
            new AbstractFunction1<List<LogSegmentMetadata>, Future<Long>>() {
                @Override
                public Future<Long> apply(List<LogSegmentMetadata> segmentList) {
                    LOG.info("Initiating Recovery For {} : {}", getFullyQualifiedName(), segmentList);
                    // if lastLedgerRollingTimeMillis is not updated, we set it to now.
                    synchronized (BKLogWriteHandler.this) {
                        if (lastLedgerRollingTimeMillis < 0) {
                            lastLedgerRollingTimeMillis = Utils.nowInMillis();
                        }
                    }

                    if (validateLogSegmentSequenceNumber) {
                        synchronized (inprogressLSSNs) {
                            for (LogSegmentMetadata segment : segmentList) {
                                if (segment.isInProgress()) {
                                    inprogressLSSNs.addLast(segment.getLogSegmentSequenceNumber());
                                }
                            }
                        }
                    }

                    return FutureUtils.processList(segmentList, recoverLogSegmentFunction, scheduler).map(
                            GetLastTxIdFunction.INSTANCE);
                }
            };

    // Stats
    private final StatsLogger perLogStatsLogger;
    private final OpStatsLogger closeOpStats;
    private final OpStatsLogger openOpStats;
    private final OpStatsLogger recoverOpStats;
    private final OpStatsLogger deleteOpStats;

    /**
     * Construct a Bookkeeper journal manager.
     */
    BKLogWriteHandler(ZKLogMetadataForWriter logMetadata,
                      DistributedLogConfiguration conf,
                      ZooKeeperClientBuilder zkcBuilder,
                      BookKeeperClientBuilder bkcBuilder,
                      LogSegmentMetadataStore metadataStore,
                      OrderedScheduler scheduler,
                      LedgerAllocator allocator,
                      StatsLogger statsLogger,
                      StatsLogger perLogStatsLogger,
                      AlertStatsLogger alertStatsLogger,
                      String clientId,
                      int regionId,
                      PermitLimiter writeLimiter,
                      FeatureProvider featureProvider,
                      DynamicDistributedLogConfiguration dynConf,
                      DistributedLock lock /** owned by handler **/) {
        super(logMetadata, conf, zkcBuilder, bkcBuilder, metadataStore,
              scheduler, statsLogger, alertStatsLogger, null, WRITE_HANDLE_FILTER, clientId);
        this.perLogStatsLogger = perLogStatsLogger;
        this.writeLimiter = writeLimiter;
        this.featureProvider = featureProvider;
        this.dynConf = dynConf;
        this.ledgerAllocator = allocator;
        this.lock = lock;
        this.metadataUpdater = LogSegmentMetadataStoreUpdater.createMetadataUpdater(conf, metadataStore);

        ensembleSize = conf.getEnsembleSize();

        if (ensembleSize < conf.getWriteQuorumSize()) {
            writeQuorumSize = ensembleSize;
            LOG.warn("Setting write quorum size {} greater than ensemble size {}",
                conf.getWriteQuorumSize(), ensembleSize);
        } else {
            writeQuorumSize = conf.getWriteQuorumSize();
        }
        if (writeQuorumSize < conf.getAckQuorumSize()) {
            ackQuorumSize = writeQuorumSize;
            LOG.warn("Setting write ack quorum size {} greater than write quorum size {}",
                conf.getAckQuorumSize(), writeQuorumSize);
        } else {
            ackQuorumSize = conf.getAckQuorumSize();
        }

        if (conf.getEncodeRegionIDInLogSegmentMetadata()) {
            this.regionId = regionId;
        } else {
            this.regionId = DistributedLogConstants.LOCAL_REGION_ID;
        }
        this.sanityCheckTxnId = conf.getSanityCheckTxnID();
        this.validateLogSegmentSequenceNumber = conf.isLogSegmentSequenceNumberValidationEnabled();

        // Construct the max sequence no
        maxLogSegmentSequenceNo = new MaxLogSegmentSequenceNo(logMetadata.getMaxLSSNData());
        inprogressLSSNs = new LinkedList<Long>();
        // Construct the max txn id.
        maxTxId = new MaxTxId(zooKeeperClient, logMetadata.getMaxTxIdPath(),
                conf.getSanityCheckTxnID(), logMetadata.getMaxTxIdData());

        // Schedule fetching ledgers list in background before we access it.
        // We don't need to watch the ledgers list changes for writer, as it manages ledgers list.
        scheduleGetLedgersTask(false, true);

        // Initialize other parameters.
        setLastLedgerRollingTimeMillis(Utils.nowInMillis());

        // Rolling Policy
        if (conf.getLogSegmentRollingIntervalMinutes() > 0) {
            rollingPolicy = new TimeBasedRollingPolicy(conf.getLogSegmentRollingIntervalMinutes() * 60 * 1000L);
        } else {
            rollingPolicy = new SizeBasedRollingPolicy(conf.getMaxLogSegmentBytes());
        }

        // Stats
        StatsLogger segmentsStatsLogger = statsLogger.scope("segments");
        openOpStats = segmentsStatsLogger.getOpStatsLogger("open");
        closeOpStats = segmentsStatsLogger.getOpStatsLogger("close");
        recoverOpStats = segmentsStatsLogger.getOpStatsLogger("recover");
        deleteOpStats = segmentsStatsLogger.getOpStatsLogger("delete");
    }

    // Transactional operations for MaxLogSegmentSequenceNo
    void storeMaxSequenceNumber(final Transaction txn,
                                final MaxLogSegmentSequenceNo maxSeqNo,
                                final long seqNo,
                                final boolean isInprogress) {
        byte[] data = DLUtils.serializeLogSegmentSequenceNumber(seqNo);
        Op zkOp = Op.setData(logMetadata.getLogSegmentsPath(), data, maxSeqNo.getZkVersion());
        txn.addOp(new ZKVersionedSetOp(zkOp, new Transaction.OpListener<Version>() {
            @Override
            public void onCommit(Version version) {
                if (validateLogSegmentSequenceNumber) {
                    synchronized (inprogressLSSNs) {
                        if (isInprogress) {
                            inprogressLSSNs.add(seqNo);
                        } else {
                            inprogressLSSNs.removeFirst();
                        }
                    }
                }
                maxSeqNo.update((ZkVersion) version, seqNo);
            }

            @Override
            public void onAbort(Throwable t) {
                // no-op
            }
        }));
    }

    // Transactional operations for MaxTxId
    void storeMaxTxId(final ZKTransaction txn,
                      final MaxTxId maxTxId,
                      final long txId) {
        byte[] data = maxTxId.couldStore(txId);
        if (null != data) {
            Op zkOp = Op.setData(maxTxId.getZkPath(), data, -1);
            txn.addOp(new ZKVersionedSetOp(zkOp, new Transaction.OpListener<Version>() {
                @Override
                public void onCommit(Version version) {
                    maxTxId.setMaxTxId(txId);
                }

                @Override
                public void onAbort(Throwable t) {

                }
            }));
        }
    }

    // Transactional operations for logsegment
    void writeLogSegment(final ZKTransaction txn,
                         final List<ACL> acl,
                         final String inprogressSegmentName,
                         final LogSegmentMetadata metadata,
                         final String path) {
        byte[] finalisedData = metadata.getFinalisedData().getBytes(UTF_8);
        Op zkOp = Op.create(path, finalisedData, acl, CreateMode.PERSISTENT);
        txn.addOp(new ZKOp(zkOp) {
            @Override
            protected void commitOpResult(OpResult opResult) {
                addLogSegmentToCache(inprogressSegmentName, metadata);
            }

            @Override
            protected void abortOpResult(Throwable t, OpResult opResult) {
                // no-op
            }
        });
    }

    void deleteLogSegment(final ZKTransaction txn,
                          final String logSegmentName,
                          final String logSegmentPath) {
        Op zkOp = Op.delete(logSegmentPath, -1);
        txn.addOp(new ZKOp(zkOp) {
            @Override
            protected void commitOpResult(OpResult opResult) {
                removeLogSegmentFromCache(logSegmentName);
            }
            @Override
            protected void abortOpResult(Throwable t, OpResult opResult) {
                // no-op
            }
        });
    }

    /**
     * The caller could call this before any actions, which to hold the lock for
     * the write handler of its whole lifecycle. The lock will only be released
     * when closing the write handler.
     *
     * This method is useful to prevent releasing underlying zookeeper lock during
     * recovering/completing log segments. Releasing underlying zookeeper lock means
     * 1) increase latency when re-lock on starting new log segment. 2) increase the
     * possibility of a stream being re-acquired by other instances.
     *
     * @return future represents the lock result
     */
    Future<? extends DistributedLock> lockHandler() {
        if (null != lockFuture) {
            return lockFuture;
        }
        lockFuture = lock.asyncAcquire();
        return lockFuture;
    }

    Future<Void> unlockHandler() {
        if (null != lockFuture) {
            return lock.asyncClose();
        } else {
            return Future.Void();
        }
    }

    void register(Watcher watcher) {
        this.zooKeeperClient.register(watcher);
    }

    /**
     * Start a new log segment in a BookKeeper ledger.
     * First ensure that we have the write lock for this journal.
     * Then create a ledger and stream based on that ledger.
     * The ledger id is written to the inprogress znode, so that in the
     * case of a crash, a recovery process can find the ledger we were writing
     * to when we crashed.
     *
     * @param txId First transaction id to be written to the stream
     * @return
     * @throws IOException
     */
    public BKLogSegmentWriter startLogSegment(long txId) throws IOException {
        return startLogSegment(txId, false, false);
    }

    /**
     * Start a new log segment in a BookKeeper ledger.
     * First ensure that we have the write lock for this journal.
     * Then create a ledger and stream based on that ledger.
     * The ledger id is written to the inprogress znode, so that in the
     * case of a crash, a recovery process can find the ledger we were writing
     * to when we crashed.
     *
     * @param txId First transaction id to be written to the stream
     * @param bestEffort
     * @param allowMaxTxID
     *          allow using max tx id to start log segment
     * @return
     * @throws IOException
     */
    public BKLogSegmentWriter startLogSegment(long txId, boolean bestEffort, boolean allowMaxTxID)
            throws IOException {
        Stopwatch stopwatch = Stopwatch.createStarted();
        boolean success = false;
        try {
            BKLogSegmentWriter writer = doStartLogSegment(txId, bestEffort, allowMaxTxID);
            success = true;
            return writer;
        } finally {
            if (success) {
                openOpStats.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            } else {
                openOpStats.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            }
        }
    }

    protected long assignLogSegmentSequenceNumber() throws IOException {
        // For any active stream we will always make sure that there is at least one
        // active ledger (except when the stream first starts out). Therefore when we
        // see no ledger metadata for a stream, we assume that this is the first ledger
        // in the stream
        long logSegmentSeqNo = DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO;
        boolean logSegmentsFound = false;

        if (LogSegmentMetadata.supportsLogSegmentSequenceNo(conf.getDLLedgerMetadataLayoutVersion())) {
            List<LogSegmentMetadata> ledgerListDesc = getFilteredLedgerListDesc(false, false);
            Long nextLogSegmentSeqNo = DLUtils.nextLogSegmentSequenceNumber(ledgerListDesc);

            if (null == nextLogSegmentSeqNo) {
                logSegmentsFound = false;
                // we don't find last assigned log segment sequence number
                // then we start the log segment with configured FirstLogSegmentSequenceNumber.
                logSegmentSeqNo = conf.getFirstLogSegmentSequenceNumber();
            } else {
                logSegmentsFound = true;
                // latest log segment is assigned with a sequence number, start with next sequence number
                logSegmentSeqNo = nextLogSegmentSeqNo;
            }
        }

        // We only skip log segment sequence number validation only when no log segments found &
        // the maximum log segment sequence number is "UNASSIGNED".
        if (!logSegmentsFound &&
            (DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO == maxLogSegmentSequenceNo.getSequenceNumber())) {
            // no ledger seqno stored in /ledgers before
            LOG.info("No max ledger sequence number found while creating log segment {} for {}.",
                logSegmentSeqNo, getFullyQualifiedName());
        } else if (maxLogSegmentSequenceNo.getSequenceNumber() + 1 != logSegmentSeqNo) {
            LOG.warn("Unexpected max log segment sequence number {} for {} : list of cached segments = {}",
                new Object[]{maxLogSegmentSequenceNo.getSequenceNumber(), getFullyQualifiedName(),
                    getCachedLogSegments(LogSegmentMetadata.DESC_COMPARATOR)});
            // there is max log segment number recorded there and it isn't match. throw exception.
            throw new DLIllegalStateException("Unexpected max log segment sequence number "
                + maxLogSegmentSequenceNo.getSequenceNumber() + " for " + getFullyQualifiedName()
                + ", expected " + (logSegmentSeqNo - 1));
        }

        return logSegmentSeqNo;
    }

    protected BKLogSegmentWriter doStartLogSegment(long txId, boolean bestEffort, boolean allowMaxTxID) throws IOException {
        return FutureUtils.result(asyncStartLogSegment(txId, bestEffort, allowMaxTxID));
    }

    protected Future<BKLogSegmentWriter> asyncStartLogSegment(long txId,
                                                              boolean bestEffort,
                                                              boolean allowMaxTxID) {
        Promise<BKLogSegmentWriter> promise = new Promise<BKLogSegmentWriter>();
        try {
            lock.checkOwnershipAndReacquire();
        } catch (LockingException e) {
            FutureUtils.setException(promise, e);
            return promise;
        }
        doStartLogSegment(txId, bestEffort, allowMaxTxID, promise);
        return promise;
    }

    protected void doStartLogSegment(final long txId,
                                     final boolean bestEffort,
                                     final boolean allowMaxTxID,
                                     final Promise<BKLogSegmentWriter> promise) {
        // validate the tx id
        if ((txId < 0) ||
                (!allowMaxTxID && (txId == DistributedLogConstants.MAX_TXID))) {
            FutureUtils.setException(promise, new IOException("Invalid Transaction Id " + txId));
            return;
        }
        if (this.sanityCheckTxnId) {
            long highestTxIdWritten = maxTxId.get();
            if (txId < highestTxIdWritten) {
                if (highestTxIdWritten == DistributedLogConstants.MAX_TXID) {
                    LOG.error("We've already marked the stream as ended and attempting to start a new log segment");
                    FutureUtils.setException(promise, new EndOfStreamException("Writing to a stream after it has been marked as completed"));
                    return;
                }
                else {
                    LOG.error("We've already seen TxId {} the max TXId is {}", txId, highestTxIdWritten);
                    FutureUtils.setException(promise, new TransactionIdOutOfOrderException(txId, highestTxIdWritten));
                    return;
                }
            }
        }

        try {
            ledgerAllocator.allocate();
        } catch (IOException e) {
            // failed to issue an allocation request
            failStartLogSegment(promise, bestEffort, e);
            return;
        }

        // start the transaction from zookeeper
        final ZKTransaction txn = new ZKTransaction(zooKeeperClient);

        // failpoint injected before creating ledger
        try {
            FailpointUtils.checkFailPoint(FailpointUtils.FailPointName.FP_StartLogSegmentBeforeLedgerCreate);
        } catch (IOException ioe) {
            failStartLogSegment(promise, bestEffort, ioe);
            return;
        }

        ledgerAllocator.tryObtain(txn, new Transaction.OpListener<LedgerHandle>() {
            @Override
            public void onCommit(LedgerHandle lh) {
                // no-op
            }

            @Override
            public void onAbort(Throwable t) {
                // no-op
            }
        }).addEventListener(new FutureEventListener<LedgerHandle>() {

            @Override
            public void onSuccess(LedgerHandle lh) {
                // try-obtain succeed
                createInprogressLogSegment(
                        txn,
                        txId,
                        lh,
                        bestEffort,
                        promise);
            }

            @Override
            public void onFailure(Throwable cause) {
                failStartLogSegment(promise, bestEffort, cause);
            }
        });
    }

    private void failStartLogSegment(Promise<BKLogSegmentWriter> promise,
                                     boolean bestEffort,
                                     Throwable cause) {
        if (bestEffort) {
            FutureUtils.setValue(promise, null);
        } else {
            FutureUtils.setException(promise, cause);
        }
    }

    // once the ledger handle is obtained from allocator, this function should guarantee
    // either the transaction is executed or aborted. Otherwise, the ledger handle will
    // just leak from the allocation pool - hence cause "No Ledger Allocator"
    private void createInprogressLogSegment(ZKTransaction txn,
                                            final long txId,
                                            final LedgerHandle lh,
                                            boolean bestEffort,
                                            final Promise<BKLogSegmentWriter> promise) {
        final long logSegmentSeqNo;
        try {
            FailpointUtils.checkFailPoint(
                    FailpointUtils.FailPointName.FP_StartLogSegmentOnAssignLogSegmentSequenceNumber);
            logSegmentSeqNo = assignLogSegmentSequenceNumber();
        } catch (IOException e) {
            // abort the current prepared transaction
            txn.abort(e);
            failStartLogSegment(promise, bestEffort, e);
            return;
        }

        final String inprogressZnodeName = inprogressZNodeName(lh.getId(), txId, logSegmentSeqNo);
        final String inprogressZnodePath = inprogressZNode(lh.getId(), txId, logSegmentSeqNo);
        final LogSegmentMetadata l =
            new LogSegmentMetadata.LogSegmentMetadataBuilder(inprogressZnodePath,
                conf.getDLLedgerMetadataLayoutVersion(), lh.getId(), txId)
                    .setLogSegmentSequenceNo(logSegmentSeqNo)
                    .setRegionId(regionId)
                    .setEnvelopeEntries(LogSegmentMetadata.supportsEnvelopedEntries(conf.getDLLedgerMetadataLayoutVersion()))
                    .build();

        // Create an inprogress segment
        writeLogSegment(
                txn,
                zooKeeperClient.getDefaultACL(),
                inprogressZnodeName,
                l,
                inprogressZnodePath);

        // Try storing max sequence number.
        LOG.debug("Try storing max sequence number in startLogSegment {} : {}", inprogressZnodePath, logSegmentSeqNo);
        storeMaxSequenceNumber(txn, maxLogSegmentSequenceNo, logSegmentSeqNo, true);

        // Try storing max tx id.
        LOG.debug("Try storing MaxTxId in startLogSegment  {} {}", inprogressZnodePath, txId);
        storeMaxTxId(txn, maxTxId, txId);

        txn.execute().addEventListener(FutureEventListenerRunnable.of(new FutureEventListener<Void>() {

            @Override
            public void onSuccess(Void value) {
                try {
                    FutureUtils.setValue(promise, new BKLogSegmentWriter(
                            getFullyQualifiedName(),
                            inprogressZnodeName,
                            conf,
                            conf.getDLLedgerMetadataLayoutVersion(),
                            new BKLogSegmentEntryWriter(lh),
                            lock,
                            txId,
                            logSegmentSeqNo,
                            scheduler,
                            statsLogger,
                            perLogStatsLogger,
                            alertStatsLogger,
                            writeLimiter,
                            featureProvider,
                            dynConf));
                } catch (IOException ioe) {
                    failStartLogSegment(promise, false, ioe);
                }
            }

            @Override
            public void onFailure(Throwable cause) {
                failStartLogSegment(promise, false, cause);
            }
        }, scheduler));
    }

    boolean shouldStartNewSegment(BKLogSegmentWriter writer) {
        return rollingPolicy.shouldRollover(writer, lastLedgerRollingTimeMillis);
    }

    /**
     * Finalize a log segment. If the journal manager is currently
     * writing to a ledger, ensure that this is the ledger of the log segment
     * being finalized.
     * <p/>
     * Otherwise this is the recovery case. In the recovery case, ensure that
     * the firstTxId of the ledger matches firstTxId for the segment we are
     * trying to finalize.
     */
    Future<LogSegmentMetadata> completeAndCloseLogSegment(final BKLogSegmentWriter writer) {
        final Promise<LogSegmentMetadata> promise = new Promise<LogSegmentMetadata>();
        completeAndCloseLogSegment(writer, promise);
        return promise;
    }

    private void completeAndCloseLogSegment(final BKLogSegmentWriter writer,
                                            final Promise<LogSegmentMetadata> promise) {
        writer.asyncClose().addEventListener(new FutureEventListener<Void>() {
            @Override
            public void onSuccess(Void value) {
                // in theory closeToFinalize should throw exception if a stream is in error.
                // just in case, add another checking here to make sure we don't close log segment is a stream is in error.
                if (writer.shouldFailCompleteLogSegment()) {
                    FutureUtils.setException(promise,
                            new IOException("LogSegmentWriter for " + writer.getFullyQualifiedLogSegment() + " is already in error."));
                    return;
                }
                doCompleteAndCloseLogSegment(
                        inprogressZNodeName(writer.getLogSegmentId(), writer.getStartTxId(), writer.getLogSegmentSequenceNumber()),
                        writer.getLogSegmentSequenceNumber(),
                        writer.getLogSegmentId(),
                        writer.getStartTxId(),
                        writer.getLastTxId(),
                        writer.getPositionWithinLogSegment(),
                        writer.getLastDLSN().getEntryId(),
                        writer.getLastDLSN().getSlotId(),
                        promise);
            }

            @Override
            public void onFailure(Throwable cause) {
                FutureUtils.setException(promise, cause);
            }
        });
    }

    @VisibleForTesting
    LogSegmentMetadata completeAndCloseLogSegment(long logSegmentSeqNo,
                                                  long ledgerId,
                                                  long firstTxId,
                                                  long lastTxId,
                                                  int recordCount)
        throws IOException {
        return completeAndCloseLogSegment(inprogressZNodeName(ledgerId, firstTxId, logSegmentSeqNo), logSegmentSeqNo,
            ledgerId, firstTxId, lastTxId, recordCount, -1, -1);
    }

    /**
     * Finalize a log segment. If the journal manager is currently
     * writing to a ledger, ensure that this is the ledger of the log segment
     * being finalized.
     * <p/>
     * Otherwise this is the recovery case. In the recovery case, ensure that
     * the firstTxId of the ledger matches firstTxId for the segment we are
     * trying to finalize.
     */
    LogSegmentMetadata completeAndCloseLogSegment(String inprogressZnodeName, long logSegmentSeqNo,
                                                  long ledgerId, long firstTxId, long lastTxId,
                                                  int recordCount, long lastEntryId, long lastSlotId)
            throws IOException {
        Stopwatch stopwatch = Stopwatch.createStarted();
        boolean success = false;
        try {
            LogSegmentMetadata completedLogSegment =
                    doCompleteAndCloseLogSegment(inprogressZnodeName, logSegmentSeqNo,
                            ledgerId, firstTxId, lastTxId, recordCount,
                            lastEntryId, lastSlotId);
            success = true;
            return completedLogSegment;
        } finally {
            if (success) {
                closeOpStats.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            } else {
                closeOpStats.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            }
        }
    }

    protected long computeStartSequenceId(LogSegmentMetadata segment) throws IOException {
        if (!segment.isInProgress()) {
            return segment.getStartSequenceId();
        }

        long startSequenceId = DistributedLogConstants.UNASSIGNED_SEQUENCE_ID;

        // we only record sequence id when both write version and logsegment's version support sequence id
