/**
 * 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.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.twitter.distributedlog.callback.LogSegmentListener;
import com.twitter.distributedlog.exceptions.DLInterruptedException;
import com.twitter.distributedlog.exceptions.LogEmptyException;
import com.twitter.distributedlog.exceptions.LogNotFoundException;
import com.twitter.distributedlog.exceptions.MetadataException;
import com.twitter.distributedlog.exceptions.UnexpectedException;
import com.twitter.distributedlog.exceptions.ZKException;
import com.twitter.distributedlog.impl.metadata.ZKLogMetadata;
import com.twitter.distributedlog.io.AsyncAbortable;
import com.twitter.distributedlog.io.AsyncCloseable;
import com.twitter.distributedlog.logsegment.LogSegmentCache;
import com.twitter.distributedlog.logsegment.LogSegmentFilter;
import com.twitter.distributedlog.logsegment.LogSegmentMetadataStore;
import com.twitter.distributedlog.util.FutureUtils;
import com.twitter.distributedlog.util.OrderedScheduler;
import com.twitter.distributedlog.util.Utils;
import com.twitter.util.Function;
import com.twitter.util.Future;
import com.twitter.util.FutureEventListener;
import com.twitter.util.Promise;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.runtime.AbstractFunction0;
import scala.runtime.BoxedUnit;

import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
 * The base class about log handler on managing log segments.
 *
 * <h3>Metrics</h3>
 * The log handler is a base class on managing log segments. so all the metrics
 * here are related to log segments retrieval and exposed under `logsegments`.
 * These metrics are all OpStats, in the format of <code>`scope`/logsegments/`op`</code>.
 * <p>
 * Those operations are:
 * <ul>
 * <li>force_get_list: force to get the list of log segments.
 * <li>get_list: get the list of the log segments. it might just retrieve from
 * local log segment cache.
 * <li>get_filtered_list: get the filtered list of log segments.
 * <li>get_full_list: get the full list of log segments.
 * <li>get_inprogress_segment: time between the inprogress log segment created and
 * the handler read it.
 * <li>get_completed_segment: time between a log segment is turned to completed and
 * the handler read it.
 * <li>negative_get_inprogress_segment: record the negative values for `get_inprogress_segment`.
 * <li>negative_get_completed_segment: record the negative values for `get_completed_segment`.
 * <li>recover_last_entry: recovering last entry from a log segment
 * <li>recover_scanned_entries: the number of entries that are scanned during recovering.
 * </ul>
 * @see BKLogWriteHandler
 * @see BKLogReadHandler
 */
public abstract class BKLogHandler implements Watcher, AsyncCloseable, AsyncAbortable {
    static final Logger LOG = LoggerFactory.getLogger(BKLogHandler.class);

    private static final int LAYOUT_VERSION = -1;

    protected final ZKLogMetadata logMetadata;
    protected final DistributedLogConfiguration conf;
    protected final ZooKeeperClient zooKeeperClient;
    protected final BookKeeperClient bookKeeperClient;
    protected final LogSegmentMetadataStore metadataStore;
    protected final int firstNumEntriesPerReadLastRecordScan;
    protected final int maxNumEntriesPerReadLastRecordScan;
    protected volatile long lastLedgerRollingTimeMillis = -1;
    protected final OrderedScheduler scheduler;
    protected final StatsLogger statsLogger;
    protected final AlertStatsLogger alertStatsLogger;
    private final AtomicBoolean ledgerListWatchSet = new AtomicBoolean(false);
    private final AtomicBoolean isFullListFetched = new AtomicBoolean(false);
    protected volatile boolean reportGetSegmentStats = false;
    private final String lockClientId;
    protected final AtomicReference<IOException> metadataException = new AtomicReference<IOException>(null);

    // listener
    protected final CopyOnWriteArraySet<LogSegmentListener> listeners =
            new CopyOnWriteArraySet<LogSegmentListener>();

    // Maintain the list of ledgers
    protected final LogSegmentCache logSegmentCache;
    protected volatile SyncGetLedgersCallback firstGetLedgersTask = null;

    protected final AsyncNotification notification;
    // log segment filter
    protected final LogSegmentFilter filter;

    // zookeeper children watcher
    private final Watcher getChildrenWatcher;

    // trace
    protected final long metadataLatencyWarnThresholdMillis;

    // Stats
    private final OpStatsLogger forceGetListStat;
    private final OpStatsLogger getListStat;
    private final OpStatsLogger getFilteredListStat;
    private final OpStatsLogger getFullListStat;
    private final OpStatsLogger getInprogressSegmentStat;
    private final OpStatsLogger getCompletedSegmentStat;
    private final OpStatsLogger negativeGetInprogressSegmentStat;
    private final OpStatsLogger negativeGetCompletedSegmentStat;
    private final OpStatsLogger recoverLastEntryStats;
    private final OpStatsLogger recoverScannedEntriesStats;

    static class SyncGetLedgersCallback implements GenericCallback<List<LogSegmentMetadata>> {

        final String path;
        final boolean allowEmpty;
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        final Promise<List<LogSegmentMetadata>> promise =
                new Promise<List<LogSegmentMetadata>>();

        int rc = KeeperException.Code.APIERROR.intValue();

        SyncGetLedgersCallback(String path, boolean allowEmpty) {
            this.path = path;
            this.allowEmpty = allowEmpty;
        }

        @Override
        public void operationComplete(int rc, List<LogSegmentMetadata> logSegmentMetadatas) {
            this.rc = rc;
            if (KeeperException.Code.OK.intValue() == rc) {
                LOG.debug("Updated ledgers list for {} : {}", path, logSegmentMetadatas);
                promise.setValue(logSegmentMetadatas);
            } else if (KeeperException.Code.NONODE.intValue() == rc) {
                if (allowEmpty) {
                    promise.setValue(new ArrayList<LogSegmentMetadata>(0));
                } else {
                    promise.setException(new LogNotFoundException("Log " + path + " is not found"));
                }
            } else {
                promise.setException(new MetadataException("Error getting ledgers list for " + path));
            }
            countDownLatch.countDown();
        }

        void waitForFinish() throws IOException {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                throw new DLInterruptedException("Interrupted on getting ledgers list for " + path, e);
            }
            if (KeeperException.Code.OK.intValue() != rc) {
                if (KeeperException.Code.NONODE.intValue() == rc) {
                    if (!allowEmpty) {
                        throw new LogNotFoundException("Log " + path + " is not found");
                    }
                } else {
                    throw new MetadataException("Error getting ledgers list for " + path);
                }
            }
        }
    }

    static class NOPGetLedgersCallback implements GenericCallback<List<LogSegmentMetadata>> {

        final String path;

        NOPGetLedgersCallback(String path) {
            this.path = path;
        }

        @Override
        public void operationComplete(int rc, List<LogSegmentMetadata> logSegmentMetadatas) {
            if (KeeperException.Code.OK.intValue() == rc) {
                LOG.debug("Updated ledgers list : {}", path, logSegmentMetadatas);
            }
        }
    }

    class WatcherGetLedgersCallback implements GenericCallback<List<LogSegmentMetadata>>, Runnable {

        final String path;

        WatcherGetLedgersCallback(String path) {
            this.path = path;
        }

        @Override
        public void operationComplete(int rc, List<LogSegmentMetadata> logSegmentMetadatas) {
            if (KeeperException.Code.OK.intValue() == rc) {
                LOG.debug("Updated ledgers list {} : {}", path, logSegmentMetadatas);
            } else {
                scheduler.schedule(this, conf.getZKRetryBackoffMaxMillis(), TimeUnit.MILLISECONDS);
            }
        }

        @Override
        public void run() {
            asyncGetLedgerListWithRetries(LogSegmentMetadata.COMPARATOR, filter, getChildrenWatcher, this);
        }
    }

    /**
     * Construct a Bookkeeper journal manager.
     */
    BKLogHandler(ZKLogMetadata metadata,
                 DistributedLogConfiguration conf,
                 ZooKeeperClientBuilder zkcBuilder,
                 BookKeeperClientBuilder bkcBuilder,
                 LogSegmentMetadataStore metadataStore,
                 OrderedScheduler scheduler,
                 StatsLogger statsLogger,
                 AlertStatsLogger alertStatsLogger,
                 AsyncNotification notification,
                 LogSegmentFilter filter,
                 String lockClientId) {
        Preconditions.checkNotNull(zkcBuilder);
        Preconditions.checkNotNull(bkcBuilder);
        this.logMetadata = metadata;
        this.conf = conf;
        this.scheduler = scheduler;
        this.statsLogger = statsLogger;
        this.alertStatsLogger = alertStatsLogger;
        this.notification = notification;
        this.filter = filter;
        this.logSegmentCache = new LogSegmentCache(metadata.getLogName());

        firstNumEntriesPerReadLastRecordScan = conf.getFirstNumEntriesPerReadLastRecordScan();
        maxNumEntriesPerReadLastRecordScan = conf.getMaxNumEntriesPerReadLastRecordScan();
        this.zooKeeperClient = zkcBuilder.build();
        LOG.debug("Using ZK Path {}", logMetadata.getLogRootPath());
        this.bookKeeperClient = bkcBuilder.build();
        this.metadataStore = metadataStore;

        if (lockClientId.equals(DistributedLogConstants.UNKNOWN_CLIENT_ID)) {
            this.lockClientId = getHostIpLockClientId();
        } else {
            this.lockClientId = lockClientId;
        }

        this.getChildrenWatcher = this.zooKeeperClient.getWatcherManager()
                .registerChildWatcher(logMetadata.getLogSegmentsPath(), this);

        // Traces
        this.metadataLatencyWarnThresholdMillis = conf.getMetadataLatencyWarnThresholdMillis();

        // Stats
        StatsLogger segmentsLogger = statsLogger.scope("logsegments");
        forceGetListStat = segmentsLogger.getOpStatsLogger("force_get_list");
        getListStat = segmentsLogger.getOpStatsLogger("get_list");
        getFilteredListStat = segmentsLogger.getOpStatsLogger("get_filtered_list");
        getFullListStat = segmentsLogger.getOpStatsLogger("get_full_list");
        getInprogressSegmentStat = segmentsLogger.getOpStatsLogger("get_inprogress_segment");
        getCompletedSegmentStat = segmentsLogger.getOpStatsLogger("get_completed_segment");
        negativeGetInprogressSegmentStat = segmentsLogger.getOpStatsLogger("negative_get_inprogress_segment");
        negativeGetCompletedSegmentStat = segmentsLogger.getOpStatsLogger("negative_get_completed_segment");
        recoverLastEntryStats = segmentsLogger.getOpStatsLogger("recover_last_entry");
        recoverScannedEntriesStats = segmentsLogger.getOpStatsLogger("recover_scanned_entries");
    }

    BKLogHandler checkMetadataException() throws IOException {
        if (null != metadataException.get()) {
            throw metadataException.get();
        }
        return this;
    }

    public void reportGetSegmentStats(boolean enabled) {
        this.reportGetSegmentStats = enabled;
    }

    public String getLockClientId() {
        return lockClientId;
    }

    private String getHostIpLockClientId() {
        try {
            return InetAddress.getLocalHost().toString();
        } catch(Exception ex) {
            return DistributedLogConstants.UNKNOWN_CLIENT_ID;
        }
    }

    protected void registerListener(LogSegmentListener listener) {
        listeners.add(listener);
    }

    protected void unregisterListener(LogSegmentListener listener) {
        listeners.remove(listener);
    }

    protected void notifyUpdatedLogSegments(List<LogSegmentMetadata> segments) {
        for (LogSegmentListener listener : listeners) {
            List<LogSegmentMetadata> listToReturn =
                    new ArrayList<LogSegmentMetadata>(segments);
            Collections.sort(listToReturn, LogSegmentMetadata.DESC_COMPARATOR);
            listener.onSegmentsUpdated(listToReturn);
        }
    }

    protected void scheduleGetAllLedgersTaskIfNeeded() {
        if (isFullListFetched.get()) {
            return;
        }
        asyncGetLedgerListWithRetries(LogSegmentMetadata.COMPARATOR, LogSegmentFilter.DEFAULT_FILTER,
                null, new NOPGetLedgersCallback(getFullyQualifiedName()));
    }

    protected void scheduleGetLedgersTask(boolean watch, boolean allowEmpty) {
        if (!watch) {
            ledgerListWatchSet.set(true);
        }
        LOG.info("Scheduling get ledgers task for {}, watch = {}.", getFullyQualifiedName(), watch);
        firstGetLedgersTask = new SyncGetLedgersCallback(getFullyQualifiedName(), allowEmpty);
        asyncGetLedgerListWithRetries(LogSegmentMetadata.COMPARATOR, filter,
                watch ? getChildrenWatcher : null, firstGetLedgersTask);
        LOG.info("Scheduled get ledgers task for {}, watch = {}.", getFullyQualifiedName(), watch);
    }

    protected void waitFirstGetLedgersTaskToFinish() throws IOException {
        SyncGetLedgersCallback task = firstGetLedgersTask;
        if (null != task) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Wait first getting ledgers task to finish for {}.", getFullyQualifiedName());
            }
            task.waitForFinish();
        }
    }

    public Future<LogRecordWithDLSN> asyncGetFirstLogRecord() {
        final Promise<LogRecordWithDLSN> promise = new Promise<LogRecordWithDLSN>();
        checkLogStreamExistsAsync().addEventListener(new FutureEventListener<Void>() {
            @Override
            public void onSuccess(Void value) {
                asyncGetFullLedgerList(true, true).addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {

                    @Override
                    public void onSuccess(List<LogSegmentMetadata> ledgerList) {
                        if (ledgerList.isEmpty()) {
                            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                            return;
                        }
                        Future<LogRecordWithDLSN> firstRecord = null;
                        for (LogSegmentMetadata ledger : ledgerList) {
                            if (!ledger.isTruncated() && (ledger.getRecordCount() > 0 || ledger.isInProgress())) {
                                firstRecord = asyncReadFirstUserRecord(ledger, DLSN.InitialDLSN);
                                break;
                            }
                        }
                        if (null != firstRecord) {
                            promise.become(firstRecord);
                        } else {
                            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                        }
                    }

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

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

    public Future<LogRecordWithDLSN> getLastLogRecordAsync(final boolean recover, final boolean includeEndOfStream) {
        final Promise<LogRecordWithDLSN> promise = new Promise<LogRecordWithDLSN>();
        checkLogStreamExistsAsync().addEventListener(new FutureEventListener<Void>() {
            @Override
            public void onSuccess(Void value) {
                asyncGetFullLedgerListDesc(true, true).addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {

                    @Override
                    public void onSuccess(List<LogSegmentMetadata> ledgerList) {
                        if (ledgerList.isEmpty()) {
                            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                            return;
                        }
                        asyncGetLastLogRecord(ledgerList.iterator(), promise, recover, false, includeEndOfStream);
                    }

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

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

    private void asyncGetLastLogRecord(final Iterator<LogSegmentMetadata> ledgerIter,
                                       final Promise<LogRecordWithDLSN> promise,
                                       final boolean fence,
                                       final boolean includeControlRecord,
                                       final boolean includeEndOfStream) {
        if (ledgerIter.hasNext()) {
            LogSegmentMetadata metadata = ledgerIter.next();
            asyncReadLastRecord(metadata, fence, includeControlRecord, includeEndOfStream).addEventListener(
                    new FutureEventListener<LogRecordWithDLSN>() {
                        @Override
                        public void onSuccess(LogRecordWithDLSN record) {
                            if (null == record) {
                                asyncGetLastLogRecord(ledgerIter, promise, fence, includeControlRecord, includeEndOfStream);
                            } else {
                                promise.setValue(record);
                            }
                        }

                        @Override
                        public void onFailure(Throwable cause) {
                            promise.setException(cause);
                        }
                    }
            );
        } else {
            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
        }
    }

    public LogRecordWithDLSN getLastLogRecord(boolean recover, boolean includeEndOfStream) throws IOException {
        checkLogStreamExists();
        List<LogSegmentMetadata> ledgerList = getFullLedgerListDesc(true, true);

        for (LogSegmentMetadata metadata: ledgerList) {
            LogRecordWithDLSN record = recoverLastRecordInLedger(metadata, recover, false, includeEndOfStream);

            if (null != record) {
                assert(!record.isControl());
                LOG.debug("{} getLastLogRecord Returned {}", getFullyQualifiedName(), record);
                return record;
            }
        }

        throw new LogEmptyException("Log " + getFullyQualifiedName() + " has no records");
    }

    public long getLastTxId(boolean recover,
                            boolean includeEndOfStream) throws IOException {
        checkLogStreamExists();
        return getLastLogRecord(recover, includeEndOfStream).getTransactionId();
    }

    public DLSN getLastDLSN(boolean recover,
                            boolean includeEndOfStream) throws IOException {
        checkLogStreamExists();
        return getLastLogRecord(recover, includeEndOfStream).getDlsn();
    }

    public long getLogRecordCount() throws IOException {
        try {
            checkLogStreamExists();
        } catch (LogNotFoundException exc) {
            return 0;
        }

        List<LogSegmentMetadata> ledgerList = getFullLedgerList(true, false);
        long count = 0;
        for (LogSegmentMetadata l : ledgerList) {
            if (l.isInProgress()) {
                LogRecord record = recoverLastRecordInLedger(l, false, false, false);
                if (null != record) {
                    count += record.getLastPositionWithinLogSegment();
                }
            } else {
                count += l.getRecordCount();
            }
        }
        return count;
    }

    private Future<LogRecordWithDLSN> asyncReadFirstUserRecord(LogSegmentMetadata ledger, DLSN beginDLSN) {
        final LedgerHandleCache handleCache =
                LedgerHandleCache.newBuilder().bkc(bookKeeperClient).conf(conf).build();
        return ReadUtils.asyncReadFirstUserRecord(
                getFullyQualifiedName(),
                ledger,
                firstNumEntriesPerReadLastRecordScan,
                maxNumEntriesPerReadLastRecordScan,
                new AtomicInteger(0),
                scheduler,
                handleCache,
                beginDLSN
        ).ensure(new AbstractFunction0<BoxedUnit>() {
            @Override
            public BoxedUnit apply() {
                handleCache.clear();
                return BoxedUnit.UNIT;
            }
        });
    }

    /**
     * This is a helper method to compactly return the record count between two records, the first denoted by
     * beginDLSN and the second denoted by endPosition. Its up to the caller to ensure that endPosition refers to
     * position in the same ledger as beginDLSN.
     */
    private Future<Long> asyncGetLogRecordCount(LogSegmentMetadata ledger, final DLSN beginDLSN, final long endPosition) {
        return asyncReadFirstUserRecord(ledger, beginDLSN).map(new Function<LogRecordWithDLSN, Long>() {
            public Long apply(final LogRecordWithDLSN beginRecord) {
                long recordCount = 0;
                if (null != beginRecord) {
                    recordCount = endPosition + 1 - beginRecord.getLastPositionWithinLogSegment();
                }
                return recordCount;
            }
        });
    }

    /**
     * Ledger metadata tells us how many records are in each completed segment, but for the first and last segments
     * we may have to crack open the entry and count. For the first entry, we need to do so because beginDLSN may be
     * an interior entry. For the last entry, if it is inprogress, we need to recover it and find the last user
     * entry.
     */
    private Future<Long> asyncGetLogRecordCount(final LogSegmentMetadata ledger, final DLSN beginDLSN) {
        if (ledger.isInProgress() && ledger.isDLSNinThisSegment(beginDLSN)) {
            return asyncReadLastUserRecord(ledger).flatMap(new Function<LogRecordWithDLSN, Future<Long>>() {
                public Future<Long> apply(final LogRecordWithDLSN endRecord) {
                    if (null != endRecord) {
                        return asyncGetLogRecordCount(ledger, beginDLSN, endRecord.getLastPositionWithinLogSegment() /* end position */);
                    } else {
                        return Future.value((long) 0);
                    }
                }
            });
        } else if (ledger.isInProgress()) {
            return asyncReadLastUserRecord(ledger).map(new Function<LogRecordWithDLSN, Long>() {
                public Long apply(final LogRecordWithDLSN endRecord) {
                    if (null != endRecord) {
                        return (long) endRecord.getLastPositionWithinLogSegment();
                    } else {
                        return (long) 0;
                    }
                }
            });
        } else if (ledger.isDLSNinThisSegment(beginDLSN)) {
            return asyncGetLogRecordCount(ledger, beginDLSN, ledger.getRecordCount() /* end position */);
        } else {
            return Future.value((long) ledger.getRecordCount());
        }
    }

    /**
     * Get a count of records between beginDLSN and the end of the stream.
     *
     * @param beginDLSN dlsn marking the start of the range
     * @return the count of records present in the range
     */
    public Future<Long> asyncGetLogRecordCount(final DLSN beginDLSN) {

        return checkLogStreamExistsAsync().flatMap(new Function<Void, Future<Long>>() {
            public Future<Long> apply(Void done) {

                return asyncGetFullLedgerList(true, false).flatMap(new Function<List<LogSegmentMetadata>, Future<Long>>() {
                    public Future<Long> apply(List<LogSegmentMetadata> ledgerList) {

                        List<Future<Long>> futureCounts = new ArrayList<Future<Long>>(ledgerList.size());
                        for (LogSegmentMetadata ledger : ledgerList) {
                            if (ledger.getLogSegmentSequenceNumber() >= beginDLSN.getLogSegmentSequenceNo()) {
                                futureCounts.add(asyncGetLogRecordCount(ledger, beginDLSN));
                            }
                        }
                        return Future.collect(futureCounts).map(new Function<List<Long>, Long>() {
                            public Long apply(List<Long> counts) {
                                return sum(counts);
                            }
                        });
                    }
                });
            }
        });
    }

    private Long sum(List<Long> values) {
        long sum = 0;
        for (Long value : values) {
            sum += value;
        }
        return sum;
    }

    public long getFirstTxId() throws IOException {
        checkLogStreamExists();
        List<LogSegmentMetadata> ledgerList = getFullLedgerList(true, true);

        // The ledger list should at least have one element
        // First TxId is populated even for in progress ledgers
        return ledgerList.get(0).getFirstTxId();
    }

    Future<Void> checkLogStreamExistsAsync() {
        final Promise<Void> promise = new Promise<Void>();
        try {
            final ZooKeeper zk = zooKeeperClient.get();
            zk.sync(logMetadata.getLogSegmentsPath(), new AsyncCallback.VoidCallback() {
                @Override
                public void processResult(int syncRc, String path, Object syncCtx) {
                    if (KeeperException.Code.NONODE.intValue() == syncRc) {
                        promise.setException(new LogNotFoundException(
                                String.format("Log %s does not exist or has been deleted", getFullyQualifiedName())));
                        return;
                    } else if (KeeperException.Code.OK.intValue() != syncRc){
                        promise.setException(new ZKException("Error on checking log existence for " + getFullyQualifiedName(),
                                KeeperException.create(KeeperException.Code.get(syncRc))));
                        return;
                    }
                    zk.exists(logMetadata.getLogSegmentsPath(), false, new AsyncCallback.StatCallback() {
                        @Override
                        public void processResult(int rc, String path, Object ctx, Stat stat) {
                            if (KeeperException.Code.OK.intValue() == rc) {
                                promise.setValue(null);
                            } else if (KeeperException.Code.NONODE.intValue() == rc) {
                                promise.setException(new LogNotFoundException(String.format("Log %s does not exist or has been deleted", getFullyQualifiedName())));
                            } else {
                                promise.setException(new ZKException("Error on checking log existence for " + getFullyQualifiedName(),
                                        KeeperException.create(KeeperException.Code.get(rc))));
                            }
                        }
                    }, null);
                }
            }, null);

        } catch (InterruptedException ie) {
            LOG.error("Interrupted while reading {}", logMetadata.getLogSegmentsPath(), ie);
            promise.setException(new DLInterruptedException("Interrupted while checking "
                    + logMetadata.getLogSegmentsPath(), ie));
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            promise.setException(e);
        }
        return promise;
    }

    private void checkLogStreamExists() throws IOException {
        try {
            if (null == Utils.sync(zooKeeperClient, logMetadata.getLogSegmentsPath())
                    .exists(logMetadata.getLogSegmentsPath(), false)) {
                throw new LogNotFoundException("Log " + getFullyQualifiedName() + " doesn't exist");
            }
        } catch (InterruptedException ie) {
            LOG.error("Interrupted while reading {}", logMetadata.getLogSegmentsPath(), ie);
            throw new DLInterruptedException("Interrupted while checking "
                    + logMetadata.getLogSegmentsPath(), ie);
        } catch (KeeperException ke) {
            LOG.error("Error checking existence for {} : ", logMetadata.getLogSegmentsPath(), ke);
            throw new ZKException("Error checking existence for " + getFullyQualifiedName() + " : ", ke);
        }
    }

    @Override
    public Future<Void> asyncClose() {
        // No-op
        this.zooKeeperClient.getWatcherManager().unregisterChildWatcher(logMetadata.getLogSegmentsPath(), this);
        return Future.Void();
    }

    @Override
    public Future<Void> asyncAbort() {
        return asyncClose();
    }

    /**
     * Find the id of the last edit log transaction written to a edit log
     * ledger.
     */
    protected Pair<Long, DLSN> readLastTxIdInLedger(LogSegmentMetadata l) throws IOException {
        LogRecordWithDLSN record = recoverLastRecordInLedger(l, false, false, true);

        if (null == record) {
            return Pair.of(DistributedLogConstants.EMPTY_LOGSEGMENT_TX_ID, DLSN.InvalidDLSN);
        }
        else {
            return Pair.of(record.getTransactionId(), record.getDlsn());
        }
    }

    /**
     * Find the id of the last edit log transaction written to a edit log
     * ledger.
     */
    protected LogRecordWithDLSN recoverLastRecordInLedger(LogSegmentMetadata l,
                                                          boolean fence,
                                                          boolean includeControl,
                                                          boolean includeEndOfStream)
        throws IOException {
        return FutureUtils.result(asyncReadLastRecord(l, fence, includeControl, includeEndOfStream));
    }

    public Future<LogRecordWithDLSN> asyncReadLastUserRecord(final LogSegmentMetadata l) {
        return asyncReadLastRecord(l, false, false, false);
    }

