/**
 * 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);
    }

    public Future<LogRecordWithDLSN> asyncReadLastRecord(final LogSegmentMetadata l,
                                                         final boolean fence,
                                                         final boolean includeControl,
                                                         final boolean includeEndOfStream) {
        final AtomicInteger numRecordsScanned = new AtomicInteger(0);
        final Stopwatch stopwatch = Stopwatch.createStarted();
        final LedgerHandleCache handleCache =
                LedgerHandleCache.newBuilder().bkc(bookKeeperClient).conf(conf).build();
        return ReadUtils.asyncReadLastRecord(
                getFullyQualifiedName(),
                l,
                fence,
                includeControl,
                includeEndOfStream,
                firstNumEntriesPerReadLastRecordScan,
                maxNumEntriesPerReadLastRecordScan,
                numRecordsScanned,
                scheduler,
                handleCache
        ).addEventListener(new FutureEventListener<LogRecordWithDLSN>() {
            @Override
            public void onSuccess(LogRecordWithDLSN value) {
                recoverLastEntryStats.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                recoverScannedEntriesStats.registerSuccessfulEvent(numRecordsScanned.get());
            }

            @Override
            public void onFailure(Throwable cause) {
                recoverLastEntryStats.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            }
        }).ensure(new AbstractFunction0<BoxedUnit>() {
            @Override
            public BoxedUnit apply() {
                handleCache.clear();
                return BoxedUnit.UNIT;
            }
        });
    }

    protected void setLastLedgerRollingTimeMillis(long rollingTimeMillis) {
        if (lastLedgerRollingTimeMillis < rollingTimeMillis) {
            lastLedgerRollingTimeMillis = rollingTimeMillis;
        }
    }

    public String getFullyQualifiedName() {
        return logMetadata.getFullyQualifiedName();
    }

    // Ledgers Related Functions
    // ***Note***
    // Get ledger list should go through #getCachedLogSegments as we need to assign start sequence id for inprogress log
    // segment so the reader could generate the right sequence id.

    protected List<LogSegmentMetadata> getCachedLogSegments(Comparator<LogSegmentMetadata> comparator)
        throws UnexpectedException {
        try {
            return logSegmentCache.getLogSegments(comparator);
        } catch (UnexpectedException ue) {
            // the log segments cache went wrong
            LOG.error("Unexpected exception on getting log segments from the cache for stream {}",
                    getFullyQualifiedName(), ue);
            metadataException.compareAndSet(null, ue);
            throw ue;
        }
    }

    protected List<LogSegmentMetadata> getFullLedgerList(boolean forceFetch, boolean throwOnEmpty)
            throws IOException {
        return getLedgerList(forceFetch, true, LogSegmentMetadata.COMPARATOR, throwOnEmpty);
    }

    protected List<LogSegmentMetadata> getFullLedgerListDesc(boolean forceFetch, boolean throwOnEmpty)
            throws IOException {
        return getLedgerList(forceFetch, true, LogSegmentMetadata.DESC_COMPARATOR, throwOnEmpty);
    }

    protected List<LogSegmentMetadata> getFilteredLedgerList(boolean forceFetch, boolean throwOnEmpty)
            throws IOException {
        return getLedgerList(forceFetch, false, LogSegmentMetadata.COMPARATOR, throwOnEmpty);
    }

    protected List<LogSegmentMetadata> getFilteredLedgerListDesc(boolean forceFetch, boolean throwOnEmpty)
            throws IOException {
        return getLedgerList(forceFetch, false, LogSegmentMetadata.DESC_COMPARATOR, throwOnEmpty);
    }

    protected List<LogSegmentMetadata> getLedgerList(boolean forceFetch,
                                                     boolean fetchFullList,
                                                     Comparator<LogSegmentMetadata> comparator,
                                                     boolean throwOnEmpty)
            throws IOException {
        Stopwatch stopwatch = Stopwatch.createStarted();
        boolean success = false;
        try {
            List<LogSegmentMetadata> segments =
                    doGetLedgerList(forceFetch, fetchFullList, comparator, throwOnEmpty);
            success = true;
            return segments;
        } finally {
            OpStatsLogger statsLogger = fetchFullList ? getFullListStat : getFilteredListStat;
            if (success) {
                statsLogger.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            } else {
                statsLogger.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            }
        }
    }

    private List<LogSegmentMetadata> doGetLedgerList(boolean forceFetch, boolean fetchFullList,
                                                     Comparator<LogSegmentMetadata> comparator,
                                                     boolean throwOnEmpty)
        throws IOException {
        if (fetchFullList) {
            if (forceFetch || !isFullListFetched.get()) {
                return forceGetLedgerList(comparator, LogSegmentFilter.DEFAULT_FILTER, throwOnEmpty);
            } else {
                return getCachedLogSegments(comparator);
            }
        } else {
            if (forceFetch) {
                return forceGetLedgerList(comparator, filter, throwOnEmpty);
            } else {
                if(!ledgerListWatchSet.get()) {
                    scheduleGetLedgersTask(true, true);
                }
                waitFirstGetLedgersTaskToFinish();
                return getCachedLogSegments(comparator);
            }
        }
    }

    /**
     * Get a list of all segments in the journal.
     */
    protected List<LogSegmentMetadata> forceGetLedgerList(final Comparator<LogSegmentMetadata> comparator,
                                                                final LogSegmentFilter segmentFilter,
                                                                boolean throwOnEmpty) throws IOException {
        final List<LogSegmentMetadata> ledgers = new ArrayList<LogSegmentMetadata>();
        final AtomicInteger result = new AtomicInteger(-1);
        final CountDownLatch latch = new CountDownLatch(1);
        Stopwatch stopwatch = Stopwatch.createStarted();
        asyncGetLedgerListInternal(comparator, segmentFilter, null, new GenericCallback<List<LogSegmentMetadata>>() {
            @Override
            public void operationComplete(int rc, List<LogSegmentMetadata> logSegmentMetadatas) {
                result.set(rc);
                if (KeeperException.Code.OK.intValue() == rc) {
                    ledgers.addAll(logSegmentMetadatas);
                } else {
                    LOG.error("Failed to get ledger list for {} : with error {}", getFullyQualifiedName(), rc);
                }
                latch.countDown();
            }
        }, new AtomicInteger(conf.getZKNumRetries()), new AtomicLong(conf.getZKRetryBackoffStartMillis()));
        try {
            latch.await();
        } catch (InterruptedException e) {
            forceGetListStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            throw new DLInterruptedException("Interrupted on reading ledger list from zkfor " + getFullyQualifiedName(), e);
        }
        long elapsedMicros = stopwatch.stop().elapsed(TimeUnit.MICROSECONDS);

        KeeperException.Code rc = KeeperException.Code.get(result.get());
        if (rc == KeeperException.Code.OK) {
            forceGetListStat.registerSuccessfulEvent(elapsedMicros);
        } else {
            forceGetListStat.registerFailedEvent(elapsedMicros);
            if (KeeperException.Code.NONODE == rc) {
                throw new LogNotFoundException("Log " + getFullyQualifiedName() + " is not found");
            } else {
                throw new IOException("ZK Exception " + rc + " reading ledger list for " + getFullyQualifiedName());
            }
        }

        if (throwOnEmpty && ledgers.isEmpty()) {
            throw new LogEmptyException("Log " + getFullyQualifiedName() + " is empty");
        }
        return ledgers;
    }

    protected Future<List<LogSegmentMetadata>> asyncGetFullLedgerList(boolean forceFetch, boolean throwOnEmpty) {
        return asyncGetLedgerList(forceFetch, true, LogSegmentMetadata.COMPARATOR, throwOnEmpty);
    }

    protected Future<List<LogSegmentMetadata>> asyncGetFullLedgerListDesc(boolean forceFetch, boolean throwOnEmpty) {
        return asyncGetLedgerList(forceFetch, true, LogSegmentMetadata.DESC_COMPARATOR, throwOnEmpty);
    }

    protected Future<List<LogSegmentMetadata>> asyncGetFilteredLedgerList(boolean forceFetch, boolean throwOnEmpty) {
        return asyncGetLedgerList(forceFetch, false, LogSegmentMetadata.COMPARATOR, throwOnEmpty);
    }

    protected Future<List<LogSegmentMetadata>> asyncGetFilteredLedgerListDesc(boolean forceFetch, boolean throwOnEmpty) {
        return asyncGetLedgerList(forceFetch, false, LogSegmentMetadata.DESC_COMPARATOR, throwOnEmpty);
    }

    protected Future<List<LogSegmentMetadata>> asyncGetLedgerList(final boolean forceFetch,
                                                                        final boolean fetchFullList,
                                                                        final Comparator<LogSegmentMetadata> comparator,
                                                                        final boolean throwOnEmpty) {
        final Promise<List<LogSegmentMetadata>> promise = new Promise<List<LogSegmentMetadata>>();
        final Stopwatch stopwatch = Stopwatch.createStarted();
        final OpStatsLogger statsLogger = fetchFullList ? getFullListStat : getFilteredListStat;
        asyncDoGetLedgerList(forceFetch, fetchFullList, comparator, throwOnEmpty)
                .addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {
                    @Override
                    public void onSuccess(List<LogSegmentMetadata> value) {
                        statsLogger.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                        promise.setValue(value);
                    }

                    @Override
                    public void onFailure(Throwable cause) {
                        statsLogger.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                        promise.setException(cause);
                    }
                });
        return promise;
    }

    private Future<List<LogSegmentMetadata>> asyncDoGetLedgerList(final boolean forceFetch,
                                                                  final boolean fetchFullList,
                                                                  final Comparator<LogSegmentMetadata> comparator,
                                                                  final boolean throwOnEmpty) {
        if (fetchFullList) {
            if (forceFetch || !isFullListFetched.get()) {
                return asyncForceGetLedgerList(comparator, LogSegmentFilter.DEFAULT_FILTER, throwOnEmpty);
            } else {
                try {
                    return Future.value(getCachedLogSegments(comparator));
                } catch (UnexpectedException ue) {
                    return Future.exception(ue);
                }
            }
        } else {
            if (forceFetch) {
                return asyncForceGetLedgerList(comparator, filter, throwOnEmpty);
            } else {
                final Promise<List<LogSegmentMetadata>> promise =
                        new Promise<List<LogSegmentMetadata>>();
                SyncGetLedgersCallback task = firstGetLedgersTask;
                task.promise.addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {
                    @Override
                    public void onSuccess(List<LogSegmentMetadata> value) {
                        try {
                            promise.setValue(getCachedLogSegments(comparator));
                        } catch (UnexpectedException e) {
                            promise.setException(e);
                        }
                    }

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

    protected Future<List<LogSegmentMetadata>> asyncForceGetLedgerList(final Comparator<LogSegmentMetadata> comparator,
                                                                       final LogSegmentFilter segmentFilter,
                                                                       final boolean throwOnEmpty) {
        final Promise<List<LogSegmentMetadata>> promise = new Promise<List<LogSegmentMetadata>>();
        final Stopwatch stopwatch = Stopwatch.createStarted();
        asyncGetLedgerListWithRetries(comparator, segmentFilter, null)
            .addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {

                @Override
                public void onSuccess(List<LogSegmentMetadata> ledgers) {
                    forceGetListStat.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                    if (ledgers.isEmpty() && throwOnEmpty) {
                        promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " is empty"));
                    } else {
                        promise.setValue(ledgers);
                    }
                }

                @Override
                public void onFailure(Throwable cause) {
                    forceGetListStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                    promise.setException(cause);
                }
            });
        return promise;
    }

    /**
     * Add the segment <i>metadata</i> for <i>name</i> in the cache.
     *
     * @param name
     *          segment znode name.
     * @param metadata
     *          segment metadata.
     */
    protected void addLogSegmentToCache(String name, LogSegmentMetadata metadata) {
        logSegmentCache.add(name, metadata);
        // update the last ledger rolling time
        if (!metadata.isInProgress() && (lastLedgerRollingTimeMillis < metadata.getCompletionTime())) {
            lastLedgerRollingTimeMillis = metadata.getCompletionTime();
        }

        if (reportGetSegmentStats) {
            // update stats
            long ts = System.currentTimeMillis();
            if (metadata.isInProgress()) {
                // as we used timestamp as start tx id we could take it as start time
                // NOTE: it is a hack here.
                long elapsedMillis = ts - metadata.getFirstTxId();
                long elapsedMicroSec = TimeUnit.MILLISECONDS.toMicros(elapsedMillis);
                if (elapsedMicroSec > 0) {
                    if (elapsedMillis > metadataLatencyWarnThresholdMillis) {
                        LOG.warn("{} received inprogress log segment in {} millis: {}",
                                 new Object[] { getFullyQualifiedName(), elapsedMillis, metadata });
                    }
                    getInprogressSegmentStat.registerSuccessfulEvent(elapsedMicroSec);
                } else {
                    negativeGetInprogressSegmentStat.registerSuccessfulEvent(-elapsedMicroSec);
                }
            } else {
                long elapsedMillis = ts - metadata.getCompletionTime();
                long elapsedMicroSec = TimeUnit.MILLISECONDS.toMicros(elapsedMillis);
                if (elapsedMicroSec > 0) {
                    if (elapsedMillis > metadataLatencyWarnThresholdMillis) {
                        LOG.warn("{} received completed log segment in {} millis : {}",
                                 new Object[] { getFullyQualifiedName(), elapsedMillis, metadata });
                    }
                    getCompletedSegmentStat.registerSuccessfulEvent(elapsedMicroSec);
                } else {
                    negativeGetCompletedSegmentStat.registerSuccessfulEvent(-elapsedMicroSec);
                }
            }
        }
    }

    protected LogSegmentMetadata readLogSegmentFromCache(String name) {
        return logSegmentCache.get(name);
    }

    protected LogSegmentMetadata removeLogSegmentFromCache(String name) {
        return logSegmentCache.remove(name);
    }

    public void asyncGetLedgerList(final Comparator<LogSegmentMetadata> comparator,
                                   Watcher watcher,
                                   final GenericCallback<List<LogSegmentMetadata>> callback) {
        asyncGetLedgerListWithRetries(comparator, filter, watcher, callback);
    }

    protected Future<List<LogSegmentMetadata>> asyncGetLedgerListWithRetries(Comparator<LogSegmentMetadata> comparator,
                                                                             LogSegmentFilter segmentFilter,
                                                                             Watcher watcher) {
        final Promise<List<LogSegmentMetadata>> promise = new Promise<List<LogSegmentMetadata>>();
        asyncGetLedgerListWithRetries(comparator, segmentFilter, watcher, new GenericCallback<List<LogSegmentMetadata>>() {
            @Override
            public void operationComplete(int rc, List<LogSegmentMetadata> segments) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    promise.setValue(segments);
                } else if (KeeperException.Code.NONODE.intValue() == rc) {
                    promise.setException(new LogNotFoundException("Log " + getFullyQualifiedName() + " not found"));
                } else {
                    String errMsg = "ZK Exception " + rc + " reading ledger list for " + getFullyQualifiedName();
                    promise.setException(new ZKException(errMsg, KeeperException.Code.get(rc)));
                }
            }
        });
        return promise;
    }

    private void asyncGetLedgerListWithRetries(final Comparator<LogSegmentMetadata> comparator,
                                               final LogSegmentFilter segmentFilter,
                                               final Watcher watcher,
                                               final GenericCallback<List<LogSegmentMetadata>> finalCallback) {
        asyncGetLedgerListInternal(comparator, segmentFilter, watcher, finalCallback,
                new AtomicInteger(conf.getZKNumRetries()), new AtomicLong(conf.getZKRetryBackoffStartMillis()));
    }

    private void asyncGetLedgerListInternal(final Comparator<LogSegmentMetadata> comparator,
                                            final LogSegmentFilter segmentFilter,
                                            final Watcher watcher,
                                            final GenericCallback<List<LogSegmentMetadata>> finalCallback,
                                            final AtomicInteger numAttemptsLeft,
                                            final AtomicLong backoffMillis) {
        final Stopwatch stopwatch = Stopwatch.createStarted();
        try {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Async getting ledger list for {}.", getFullyQualifiedName());
            }
            final GenericCallback<List<LogSegmentMetadata>> callback = new GenericCallback<List<LogSegmentMetadata>>() {
                @Override
                public void operationComplete(int rc, List<LogSegmentMetadata> result) {
                    long elapsedMicros = stopwatch.stop().elapsed(TimeUnit.MICROSECONDS);
                    if (KeeperException.Code.OK.intValue() != rc) {
                        getListStat.registerFailedEvent(elapsedMicros);
                    } else {
                        if (LogSegmentFilter.DEFAULT_FILTER == segmentFilter) {
                            isFullListFetched.set(true);
                        }
                        getListStat.registerSuccessfulEvent(elapsedMicros);
                    }
                    finalCallback.operationComplete(rc, result);
                }
            };
            zooKeeperClient.get().getChildren(logMetadata.getLogSegmentsPath(), watcher, new AsyncCallback.Children2Callback() {
                @Override
                public void processResult(final int rc, final String path, final Object ctx, final List<String> children, final Stat stat) {
                    if (KeeperException.Code.OK.intValue() != rc) {

                        if ((KeeperException.Code.CONNECTIONLOSS.intValue() == rc ||
                             KeeperException.Code.SESSIONEXPIRED.intValue() == rc ||
                             KeeperException.Code.SESSIONMOVED.intValue() == rc) &&
                            numAttemptsLeft.decrementAndGet() > 0) {
                            long backoffMs = backoffMillis.get();
                            backoffMillis.set(Math.min(conf.getZKRetryBackoffMaxMillis(), 2 * backoffMs));
                            scheduler.schedule(new Runnable() {
                                @Override
                                public void run() {
                                    asyncGetLedgerListInternal(comparator, segmentFilter, watcher,
                                            finalCallback, numAttemptsLeft, backoffMillis);
                                }
                            }, backoffMs, TimeUnit.MILLISECONDS);
                            return;
                        }
                        callback.operationComplete(rc, null);
                        return;
                    }

                    if (LOG.isTraceEnabled()) {
                        LOG.trace("Got ledger list from {} : {}", logMetadata.getLogSegmentsPath(), children);
                    }

                    ledgerListWatchSet.set(true);
                    Set<String> segmentsReceived = new HashSet<String>();
                    segmentsReceived.addAll(segmentFilter.filter(children));
                    Set<String> segmentsAdded;
                    final Set<String> removedSegments = Collections.synchronizedSet(new HashSet<String>());
                    final Map<String, LogSegmentMetadata> addedSegments =
                            Collections.synchronizedMap(new HashMap<String, LogSegmentMetadata>());
                    Pair<Set<String>, Set<String>> segmentChanges = logSegmentCache.diff(segmentsReceived);
                    segmentsAdded = segmentChanges.getLeft();
                    removedSegments.addAll(segmentChanges.getRight());

                    if (segmentsAdded.isEmpty()) {
                        if (LOG.isTraceEnabled()) {
                            LOG.trace("No segments added for {}.", getFullyQualifiedName());
                        }

                        // update the cache before fetch
                        logSegmentCache.update(removedSegments, addedSegments);

                        List<LogSegmentMetadata> segmentList;
                        try {
                            segmentList = getCachedLogSegments(comparator);
                        } catch (UnexpectedException e) {
                            callback.operationComplete(KeeperException.Code.DATAINCONSISTENCY.intValue(), null);
                            return;
                        }
                        callback.operationComplete(KeeperException.Code.OK.intValue(), segmentList);
                        notifyUpdatedLogSegments(segmentList);
                        if (!removedSegments.isEmpty()) {
                            notifyOnOperationComplete();
                        }
                        return;
                    }

                    final AtomicInteger numChildren = new AtomicInteger(segmentsAdded.size());
                    final AtomicInteger numFailures = new AtomicInteger(0);
                    for (final String segment: segmentsAdded) {
                        metadataStore.getLogSegment(logMetadata.getLogSegmentPath(segment))
                                .addEventListener(new FutureEventListener<LogSegmentMetadata>() {

                                    @Override
                                    public void onSuccess(LogSegmentMetadata result) {
                                        addedSegments.put(segment, result);
                                        complete();
                                    }

                                    @Override
                                    public void onFailure(Throwable cause) {
                                        // NONODE exception is possible in two cases
                                        // 1. A log segment was deleted by truncation between the call to getChildren and read
                                        // attempt on the znode corresponding to the segment
                                        // 2. In progress segment has been completed => inprogress ZNode does not exist
                                        if (cause instanceof KeeperException &&
                                                KeeperException.Code.NONODE == ((KeeperException) cause).code()) {
                                            removedSegments.add(segment);
                                            complete();
                                        } else {
                                            // fail fast
                                            if (1 == numFailures.incrementAndGet()) {
                                                int rcToReturn = KeeperException.Code.SYSTEMERROR.intValue();
                                                if (cause instanceof KeeperException) {
                                                    rcToReturn = ((KeeperException) cause).code().intValue();
                                                } else if (cause instanceof ZKException) {
                                                    rcToReturn = ((ZKException) cause).getKeeperExceptionCode().intValue();
                                                }
                                                // :( properly we need dlog related response code.
                                                callback.operationComplete(rcToReturn, null);
                                                return;
                                            }
                                        }
                                    }

                                    private void complete() {
                                        if (0 == numChildren.decrementAndGet() && numFailures.get() == 0) {
                                            // update the cache only when fetch completed
                                            logSegmentCache.update(removedSegments, addedSegments);
                                            List<LogSegmentMetadata> segmentList;
                                            try {
                                                segmentList = getCachedLogSegments(comparator);
                                            } catch (UnexpectedException e) {
                                                callback.operationComplete(KeeperException.Code.DATAINCONSISTENCY.intValue(), null);
                                                return;
                                            }
                                            callback.operationComplete(KeeperException.Code.OK.intValue(), segmentList);
                                            notifyUpdatedLogSegments(segmentList);
                                            notifyOnOperationComplete();
                                        }
                                    }
                                });
                    }
                }
            }, null);
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            getListStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            finalCallback.operationComplete(KeeperException.Code.CONNECTIONLOSS.intValue(), null);
        } catch (InterruptedException e) {
            getListStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
            finalCallback.operationComplete(KeeperException.Code.CONNECTIONLOSS.intValue(), null);
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (Watcher.Event.EventType.None.equals(event.getType())) {
            if (event.getState() == Watcher.Event.KeeperState.Expired) {
                // if the watcher is expired
                scheduler.schedule(new WatcherGetLedgersCallback(getFullyQualifiedName()),
                        conf.getZKRetryBackoffStartMillis(), TimeUnit.MILLISECONDS);
            }
        } else if (Watcher.Event.EventType.NodeChildrenChanged.equals(event.getType())) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("LogSegments Changed under {}.", getFullyQualifiedName());
            }
            asyncGetLedgerListWithRetries(LogSegmentMetadata.COMPARATOR, filter,
                    getChildrenWatcher, new WatcherGetLedgersCallback(getFullyQualifiedName()));
        }
    }

    void notifyOnOperationComplete() {
        if (null != notification) {
            notification.notifyOnOperationComplete();
        }
    }

    // ZooKeeper Watchers

    Watcher registerExpirationHandler(final ZooKeeperClient.ZooKeeperSessionExpireNotifier onExpired) {
        if (conf.getZKNumRetries() > 0) {
            return new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    // nop
                }
            };
        }
        return zooKeeperClient.registerExpirationHandler(onExpired);
    }

    boolean unregister(Watcher watcher) {
        return zooKeeperClient.unregister(watcher);
    }
}