/**
 * 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.readahead;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import com.twitter.distributedlog.exceptions.AlreadyTruncatedTransactionException;
import com.twitter.distributedlog.AsyncNotification;
import com.twitter.distributedlog.BKLogHandler;
import com.twitter.distributedlog.DLSN;
import com.twitter.distributedlog.DistributedLogConfiguration;
import com.twitter.distributedlog.DistributedLogConstants;
import com.twitter.distributedlog.LedgerDescriptor;
import com.twitter.distributedlog.LedgerHandleCache;
import com.twitter.distributedlog.LedgerReadPosition;
import com.twitter.distributedlog.exceptions.LogNotFoundException;
import com.twitter.distributedlog.exceptions.LogReadException;
import com.twitter.distributedlog.LogSegmentMetadata;
import com.twitter.distributedlog.ReadAheadCache;
import com.twitter.distributedlog.ZooKeeperClient;
import com.twitter.distributedlog.callback.ReadAheadCallback;
import com.twitter.distributedlog.config.DynamicDistributedLogConfiguration;
import com.twitter.distributedlog.exceptions.DLInterruptedException;
import com.twitter.distributedlog.impl.metadata.ZKLogMetadataForReader;
import com.twitter.distributedlog.injector.AsyncFailureInjector;
import com.twitter.distributedlog.io.AsyncCloseable;
import com.twitter.distributedlog.stats.ReadAheadExceptionsLogger;
import com.twitter.distributedlog.util.FutureUtils;
import com.twitter.distributedlog.util.OrderedScheduler;
import com.twitter.util.Future;
import com.twitter.util.FutureEventListener;
import com.twitter.util.Promise;
import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.MathUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Function1;
import scala.runtime.AbstractFunction1;
import scala.runtime.BoxedUnit;

import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * ReadAhead Worker process readahead in asynchronous way. The whole readahead process are chained into
 * different phases:
 *
 * <p>
 * ScheduleReadAheadPhase: Schedule the readahead request based on previous state (e.g. whether to backoff).
 * After the readahead request was scheduled, the worker enters ReadAhead phase.
 * </p>
 * <p>
 * ReadAhead Phase: This phase is divided into several sub-phases. All the sub-phases are chained into the
 * execution flow. If errors happened during execution, the worker enters Exceptions Handling Phase.
 * <br>
 * CheckInProgressChangedPhase: check whether there is in progress ledgers changed and updated the metadata.
 * <br>
 * OpenLedgerPhase: open ledgers if necessary for following read requests.
 * <br>
 * ReadEntriesPhase: read entries from bookkeeper and fill the readahead cache.
 * <br>
 * After that, the worker goes back to Schedule Phase to schedule next readahead request.
 * </p>
 * <p>
 * Exceptions Handling Phase: Handle all the exceptions and properly schedule next readahead request.
 * </p>
 */
public class ReadAheadWorker implements ReadAheadCallback, Runnable, Watcher, AsyncCloseable {

    private static final Logger LOG = LoggerFactory.getLogger(ReadAheadWorker.class);

    private static final int BKC_ZK_EXCEPTION_THRESHOLD_IN_SECONDS = 30;
    private static final int BKC_UNEXPECTED_EXCEPTION_THRESHOLD = 3;

    // Stream information
    private final String fullyQualifiedName;
    private final DistributedLogConfiguration conf;
    private final DynamicDistributedLogConfiguration dynConf;
    private final ZKLogMetadataForReader logMetadata;
    private final BKLogHandler bkLedgerManager;
    private final boolean isHandleForReading;
    // Notification to notify readahead status
    protected final AsyncNotification notification;

    // resources
    private final ZooKeeperClient zkc;
    protected final OrderedScheduler scheduler;
    private final LedgerHandleCache handleCache;
    private final ReadAheadCache readAheadCache;

    // ReadAhead Status
    volatile boolean running = true;
    Promise<Void> stopPromise = null;
    private volatile boolean isCatchingUp = true;
    private volatile boolean logDeleted = false;
    private volatile boolean readAheadError = false;
    private volatile boolean readAheadInterrupted = false;
    private volatile boolean readingFromTruncated = false;

    // Exceptions Handling
    volatile boolean encounteredException = false;
    private final AtomicInteger bkcZkExceptions = new AtomicInteger(0);
    private final AtomicInteger bkcUnExpectedExceptions = new AtomicInteger(0);
    private final int noLedgerExceptionOnReadLACThreshold;
    private final AtomicInteger bkcNoLedgerExceptionsOnReadLAC = new AtomicInteger(0);

    // Read Ahead Positions
    private final LedgerReadPosition startReadPosition;
    protected LedgerReadPosition nextReadAheadPosition;

    //
    // LogSegments & Metadata Notification
    //

    // variables related to getting log segments from zookeeper
    volatile boolean zkNotificationDisabled = false;
    private final Watcher getLedgersWatcher;

    // variables related to zookeeper watcher notification to interrupt long poll waits
    final Object notificationLock = new Object();
    AsyncNotification metadataNotification = null;
    volatile long metadataNotificationTimeMillis = -1L;

    // variables related to log segments
    private volatile boolean reInitializeMetadata = true;
    volatile boolean inProgressChanged = false;
    private LogSegmentMetadata currentMetadata = null;
    private int currentMetadataIndex;
    protected LedgerDescriptor currentLH;
    private volatile List<LogSegmentMetadata> ledgerList;

    //
    // ReadAhead Phases
    //

    final Phase schedulePhase = new ScheduleReadAheadPhase();
    final Phase exceptionHandler = new ExceptionHandlePhase(schedulePhase);
    final Phase readAheadPhase =
            new StoppablePhase(
                    new CheckInProgressChangedPhase(
                            new OpenLedgerPhase(
                                    new ReadEntriesPhase(schedulePhase))));

    //
    // Stats, Tracing and Failure Injection
    //

    // failure injector
    private final AsyncFailureInjector failureInjector;
    // trace
    protected final long metadataLatencyWarnThresholdMillis;
    final ReadAheadTracker tracker;
    final Stopwatch resumeStopWatch;
    final Stopwatch lastLedgerCloseDetected = Stopwatch.createUnstarted();
    // Misc
    private final boolean readAheadSkipBrokenEntries;
    // Stats
    private final AlertStatsLogger alertStatsLogger;
    private final StatsLogger readAheadPerStreamStatsLogger;
    private final Counter readAheadWorkerWaits;
    private final Counter readAheadEntryPiggyBackHits;
    private final Counter readAheadEntryPiggyBackMisses;
    private final Counter readAheadReadLACAndEntryCounter;
    private final Counter readAheadCacheFullCounter;
    private final Counter readAheadSkippedBrokenEntries;
    private final Counter idleReaderWarn;
    private final OpStatsLogger readAheadReadEntriesStat;
    private final OpStatsLogger readAheadCacheResumeStat;
    private final OpStatsLogger readAheadLacLagStats;
    private final OpStatsLogger longPollInterruptionStat;
    private final OpStatsLogger metadataReinitializationStat;
    private final OpStatsLogger notificationExecutionStat;
    private final ReadAheadExceptionsLogger readAheadExceptionsLogger;

    public ReadAheadWorker(DistributedLogConfiguration conf,
                           DynamicDistributedLogConfiguration dynConf,
                           ZKLogMetadataForReader logMetadata,
                           BKLogHandler ledgerManager,
                           ZooKeeperClient zkc,
                           OrderedScheduler scheduler,
                           LedgerHandleCache handleCache,
                           LedgerReadPosition startPosition,
                           ReadAheadCache readAheadCache,
                           boolean isHandleForReading,
                           ReadAheadExceptionsLogger readAheadExceptionsLogger,
                           StatsLogger handlerStatsLogger,
                           StatsLogger readAheadPerStreamStatsLogger,
                           AlertStatsLogger alertStatsLogger,
                           AsyncFailureInjector failureInjector,
                           AsyncNotification notification) {
        // Log information
        this.fullyQualifiedName = logMetadata.getFullyQualifiedName();
        this.conf = conf;
        this.dynConf = dynConf;
        this.logMetadata = logMetadata;
        this.bkLedgerManager = ledgerManager;
        this.isHandleForReading = isHandleForReading;
        this.notification = notification;
        // Resources
        this.zkc = zkc;
        this.scheduler = scheduler;
        this.handleCache = handleCache;
        this.readAheadCache = readAheadCache;
        // Readahead status
        this.startReadPosition = new LedgerReadPosition(startPosition);
        this.nextReadAheadPosition = new LedgerReadPosition(startPosition);
        // LogSegments
        this.getLedgersWatcher = this.zkc.getWatcherManager()
                .registerChildWatcher(logMetadata.getLogSegmentsPath(), this);
        // Failure Detection
        this.failureInjector = failureInjector;
        // Tracing
        this.metadataLatencyWarnThresholdMillis = conf.getMetadataLatencyWarnThresholdMillis();
        this.noLedgerExceptionOnReadLACThreshold =
                conf.getReadAheadNoSuchLedgerExceptionOnReadLACErrorThresholdMillis() / conf.getReadAheadWaitTime();
        this.tracker = new ReadAheadTracker(logMetadata.getLogName(), readAheadCache,
                ReadAheadPhase.SCHEDULE_READAHEAD, readAheadPerStreamStatsLogger);
        this.resumeStopWatch = Stopwatch.createUnstarted();
        // Misc
        this.readAheadSkipBrokenEntries = conf.getReadAheadSkipBrokenEntries();
        // Stats
        this.alertStatsLogger = alertStatsLogger;
        this.readAheadPerStreamStatsLogger = readAheadPerStreamStatsLogger;
        StatsLogger readAheadStatsLogger = handlerStatsLogger.scope("readahead_worker");
        readAheadWorkerWaits = readAheadStatsLogger.getCounter("wait");
        readAheadEntryPiggyBackHits = readAheadStatsLogger.getCounter("entry_piggy_back_hits");
        readAheadEntryPiggyBackMisses = readAheadStatsLogger.getCounter("entry_piggy_back_misses");
        readAheadReadEntriesStat = readAheadStatsLogger.getOpStatsLogger("read_entries");
        readAheadReadLACAndEntryCounter = readAheadStatsLogger.getCounter("read_lac_and_entry_counter");
        readAheadCacheFullCounter = readAheadStatsLogger.getCounter("cache_full");
        readAheadSkippedBrokenEntries = readAheadStatsLogger.getCounter("skipped_broken_entries");
        readAheadCacheResumeStat = readAheadStatsLogger.getOpStatsLogger("resume");
        readAheadLacLagStats = readAheadStatsLogger.getOpStatsLogger("read_lac_lag");
        longPollInterruptionStat = readAheadStatsLogger.getOpStatsLogger("long_poll_interruption");
        notificationExecutionStat = readAheadStatsLogger.getOpStatsLogger("notification_execution");
        metadataReinitializationStat = readAheadStatsLogger.getOpStatsLogger("metadata_reinitialization");
        idleReaderWarn = readAheadStatsLogger.getCounter("idle_reader_warn");
        this.readAheadExceptionsLogger = readAheadExceptionsLogger;
    }

    @VisibleForTesting
    public LedgerReadPosition getNextReadAheadPosition() {
        return nextReadAheadPosition;
    }

    public LedgerDescriptor getCurrentLedgerDescriptor() {
        return currentLH;
    }

    //
    // ReadAhead Status
    //

    void setReadAheadError(ReadAheadTracker tracker) {
        LOG.error("Read Ahead for {} is set to error.", logMetadata.getFullyQualifiedName());
        readAheadError = true;
        tracker.enterPhase(ReadAheadPhase.ERROR);
        if (null != notification) {
            notification.notifyOnError();
        }
        if (null != stopPromise) {
            FutureUtils.setValue(stopPromise, null);
        }
    }

    void setReadAheadInterrupted(ReadAheadTracker tracker) {
        readAheadInterrupted = true;
        tracker.enterPhase(ReadAheadPhase.INTERRUPTED);
        if (null != notification) {
            notification.notifyOnError();
        }
        if (null != stopPromise) {
            FutureUtils.setValue(stopPromise, null);
        }
    }

    void setReadingFromTruncated(ReadAheadTracker tracker) {
        readingFromTruncated = true;
        tracker.enterPhase(ReadAheadPhase.TRUNCATED);
        if (null != notification) {
            notification.notifyOnError();
        }
        if (null != stopPromise) {
            FutureUtils.setValue(stopPromise, null);
        }
    }

    private void setReadAheadStopped() {
        tracker.enterPhase(ReadAheadPhase.STOPPED);
        if (null != stopPromise) {
            FutureUtils.setValue(stopPromise, null);
        }
        LOG.info("Stopped ReadAheadWorker for {}", fullyQualifiedName);
    }

    public void checkClosedOrInError()
            throws LogNotFoundException, LogReadException, DLInterruptedException,
            AlreadyTruncatedTransactionException {
        if (logDeleted) {
            throw new LogNotFoundException(logMetadata.getFullyQualifiedName() + " is already deleted.");
        } else if (readingFromTruncated) {
            throw new AlreadyTruncatedTransactionException(
                String.format("%s: Trying to position read ahead a segment that is marked truncated",
                    logMetadata.getFullyQualifiedName()));
        } else if (readAheadInterrupted) {
            throw new DLInterruptedException(String.format("%s: ReadAhead Thread was interrupted",
                logMetadata.getFullyQualifiedName()));
        } else if (readAheadError) {
            throw new LogReadException(String.format("%s: ReadAhead Thread encountered exceptions",
                logMetadata.getFullyQualifiedName()));
        }
    }

    public boolean isCaughtUp() {
        return !isCatchingUp;
    }

    public void start() {
        LOG.debug("Starting ReadAhead Worker for {}", fullyQualifiedName);
        running = true;
        schedulePhase.process(BKException.Code.OK);
    }

    @Override
    public Future<Void> asyncClose() {
        LOG.info("Stopping Readahead worker for {}", fullyQualifiedName);
        running = false;

        this.zkc.getWatcherManager()
                .unregisterChildWatcher(this.logMetadata.getLogSegmentsPath(), this);

        // Aside from unfortunate naming of variables, this allows
        // the currently active long poll to be interrupted and completed
        AsyncNotification notification;
        synchronized (notificationLock) {
            notification = metadataNotification;
            metadataNotification = null;
        }
        if (null != notification) {
            notification.notifyOnOperationComplete();
        }
        if (null == stopPromise) {
            return Future.Void();
        }
        return FutureUtils.ignore(FutureUtils.within(
                stopPromise,
                2 * conf.getReadAheadWaitTime(),
                TimeUnit.MILLISECONDS,
                new TimeoutException("Timeout on waiting for ReadAhead worker to stop " + fullyQualifiedName),
                scheduler,
                fullyQualifiedName));
    }

    @Override
    public String toString() {
        return "Running:" + running +
            ", NextReadAheadPosition:" + nextReadAheadPosition +
            ", BKZKExceptions:" + bkcZkExceptions.get() +
            ", BKUnexpectedExceptions:" + bkcUnExpectedExceptions.get() +
            ", EncounteredException:" + encounteredException +
            ", readAheadError:" + readAheadError +
            ", readAheadInterrupted" + readAheadInterrupted +
            ", CurrentMetadata:" + ((null != currentMetadata) ? currentMetadata : "NONE") +
            ", FailureInjector:" + failureInjector;
    }

    @Override
    public void resumeReadAhead() {
        try {
            long cacheResumeLatency = resumeStopWatch.stop().elapsed(TimeUnit.MICROSECONDS);
            readAheadCacheResumeStat.registerSuccessfulEvent(cacheResumeLatency);
        } catch (IllegalStateException ise) {
            LOG.error("Encountered illegal state when stopping resume stop watch for {} : ",
                    logMetadata.getFullyQualifiedName(), ise);
        }
        submit(this);
    }

    Runnable addRTEHandler(final Runnable runnable) {
        return new Runnable() {
            @Override
            public void run() {
                try {
                    runnable.run();
                } catch (RuntimeException rte) {
                    LOG.error("ReadAhead on stream {} encountered runtime exception",
                            logMetadata.getFullyQualifiedName(), rte);
                    setReadAheadError(tracker);
                    throw rte;
                }
            }
        };
    }

    <T> Function1<T, BoxedUnit> submit(final Function1<T, BoxedUnit> function) {
        return new AbstractFunction1<T, BoxedUnit>() {
            @Override
            public BoxedUnit apply(final T input) {
                submit(new Runnable() {
                    @Override
                    public void run() {
                        function.apply(input);
                    }
                });
                return BoxedUnit.UNIT;
            }
        };
    }

    void submit(Runnable runnable) {
        if (failureInjector.shouldInjectStops()) {
            LOG.warn("Error injected: read ahead for stream {} is going to stall.",
                    logMetadata.getFullyQualifiedName());
            return;
        }

        if (failureInjector.shouldInjectDelays()) {
            int delayMs = failureInjector.getInjectedDelayMs();
            schedule(runnable, delayMs);
            return;
        }

        try {
            scheduler.submit(addRTEHandler(runnable));
        } catch (RejectedExecutionException ree) {
            setReadAheadError(tracker);
        }
    }

    private void schedule(Runnable runnable, long timeInMillis) {
        try {
            InterruptibleScheduledRunnable task = new InterruptibleScheduledRunnable(runnable);
            boolean executeImmediately = setMetadataNotification(task);
            if (executeImmediately) {
                scheduler.submit(addRTEHandler(task));
                return;
            }
            scheduler.schedule(addRTEHandler(task), timeInMillis, TimeUnit.MILLISECONDS);
            readAheadWorkerWaits.inc();
        } catch (RejectedExecutionException ree) {
            setReadAheadError(tracker);
        }
    }

    private void handleException(ReadAheadPhase phase, int returnCode) {
        readAheadExceptionsLogger.getBKExceptionStatsLogger(phase.name()).getExceptionCounter(returnCode).inc();
        exceptionHandler.process(returnCode);
    }

    private boolean closeCurrentLedgerHandle() {
        if (currentLH == null) {
            return true;
        }
        boolean retVal = false;
        LedgerDescriptor ld = currentLH;
        try {
            handleCache.closeLedger(ld);
            currentLH = null;
            retVal = true;
        } catch (BKException bke) {
            LOG.debug("BK Exception during closing {} : ", ld, bke);
            handleException(ReadAheadPhase.CLOSE_LEDGER, bke.getCode());
        }

        return retVal;
    }

    abstract class Phase {

        final Phase next;

        Phase(Phase next) {
            this.next = next;
        }

        abstract void process(int rc);
    }

    /**
     * Schedule next readahead request. If we need to backoff, schedule in a backoff delay.
     */
    final class ScheduleReadAheadPhase extends Phase {

        ScheduleReadAheadPhase() {
            super(null);
        }

        @Override
        void process(int rc) {
            if (!running) {
                setReadAheadStopped();
                return;
            }
            tracker.enterPhase(ReadAheadPhase.SCHEDULE_READAHEAD);

            boolean injectErrors = failureInjector.shouldInjectErrors();
            if (encounteredException || injectErrors) {
                int zkErrorThreshold = BKC_ZK_EXCEPTION_THRESHOLD_IN_SECONDS * 1000 * 4 / conf.getReadAheadWaitTime();

                if ((bkcZkExceptions.get() > zkErrorThreshold) || injectErrors) {
                    LOG.error("{} : BookKeeper Client used by the ReadAhead Thread has encountered {} zookeeper exceptions : simulate = {}",
                              new Object[] { fullyQualifiedName, bkcZkExceptions.get(), injectErrors });
                    running = false;
                    setReadAheadError(tracker);
                } else if (bkcUnExpectedExceptions.get() > BKC_UNEXPECTED_EXCEPTION_THRESHOLD) {
                    LOG.error("{} : ReadAhead Thread has encountered {} unexpected BK exceptions.",
                              fullyQualifiedName, bkcUnExpectedExceptions.get());
                    running = false;
                    setReadAheadError(tracker);
                } else {
                    // We must always reinitialize metadata if the last attempt to read failed.
                    reInitializeMetadata = true;
                    encounteredException = false;
                    // Backoff before resuming
                    if (LOG.isTraceEnabled()) {
                        LOG.trace("Scheduling read ahead for {} after {} ms.", fullyQualifiedName, conf.getReadAheadWaitTime() / 4);
                    }
                    schedule(ReadAheadWorker.this, conf.getReadAheadWaitTime() / 4);
                }
            } else {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("Scheduling read ahead for {} now.", fullyQualifiedName);
                }
                submit(ReadAheadWorker.this);
            }
        }

    }

    /**
     * Phase on handling exceptions.
     */
    final class ExceptionHandlePhase extends Phase {

        ExceptionHandlePhase(Phase next) {
            super(next);
        }

        @Override
        void process(int rc) {
            tracker.enterPhase(ReadAheadPhase.EXCEPTION_HANDLING);

            if (BKException.Code.InterruptedException == rc) {
                LOG.trace("ReadAhead Worker for {} is interrupted.", fullyQualifiedName);
                running = false;
                setReadAheadInterrupted(tracker);
                return;
            } else if (BKException.Code.ZKException == rc) {
                encounteredException = true;
                int numExceptions = bkcZkExceptions.incrementAndGet();
                LOG.debug("ReadAhead Worker for {} encountered zookeeper exception : total exceptions are {}.",
                        fullyQualifiedName, numExceptions);
            } else if (BKException.Code.OK != rc) {
                encounteredException = true;
                switch(rc) {
                    case BKException.Code.NoSuchEntryException:
                    case BKException.Code.LedgerRecoveryException:
                    case BKException.Code.NoSuchLedgerExistsException:
                        break;
                    default:
                        bkcUnExpectedExceptions.incrementAndGet();
                }
                LOG.info("ReadAhead Worker for {} encountered exception : {}",
                        fullyQualifiedName, BKException.create(rc));
            }
            // schedule next read ahead
            next.process(BKException.Code.OK);
        }
    }

    /**
     * A phase that could be stopped by a stopPromise
     */
    final class StoppablePhase extends Phase {

        StoppablePhase(Phase next) {
            super(next);
        }

        @Override
        void process(int rc) {
            if (!running) {
                setReadAheadStopped();
                return;
            }

            if (null == stopPromise) {
                stopPromise = new Promise<Void>();
            }

            // proceed the readahead request
            next.process(BKException.Code.OK);
        }
    }

    /**
     * Phase on checking in progress changed.
     */
    final class CheckInProgressChangedPhase extends Phase
        implements BookkeeperInternalCallbacks.GenericCallback<List<LogSegmentMetadata>> {

        CheckInProgressChangedPhase(Phase next) {
            super(next);
        }

        @Override
        public void operationComplete(final int rc, final List<LogSegmentMetadata> result) {
            // submit callback execution to dlg executor to avoid deadlock.
            submit(new Runnable() {
                @Override
                public void run() {
                    if (KeeperException.Code.OK.intValue() != rc) {
                        if (KeeperException.Code.NONODE.intValue() == rc) {
                            LOG.info("Log {} has been deleted. Set ReadAhead to error to stop reading.",
                                    logMetadata.getFullyQualifiedName());
                            logDeleted = true;
                            setReadAheadError(tracker);
                            return;
                        }
                        LOG.info("ZK Exception {} while reading ledger list", rc);
                        reInitializeMetadata = true;
                        if (DistributedLogConstants.DL_INTERRUPTED_EXCEPTION_RESULT_CODE == rc) {
                            handleException(ReadAheadPhase.GET_LEDGERS, BKException.Code.InterruptedException);
                        } else {
                            handleException(ReadAheadPhase.GET_LEDGERS, BKException.Code.ZKException);
                        }
                        return;
                    }
                    ledgerList = result;
                    boolean isInitialPositioning = nextReadAheadPosition.definitelyLessThanOrEqualTo(startReadPosition);
                    for (int i = 0; i < ledgerList.size(); i++) {
                        LogSegmentMetadata l = ledgerList.get(i);
                        // By default we should skip truncated segments during initial positioning
                        if (l.isTruncated() &&
                            isInitialPositioning &&
                            !conf.getIgnoreTruncationStatus()) {
                            continue;
                        }

                        DLSN nextReadDLSN = new DLSN(nextReadAheadPosition.getLogSegmentSequenceNumber(),
                                nextReadAheadPosition.getEntryId(), -1);

                        // next read position still inside a log segment
                        final boolean hasDataToRead = (l.getLastDLSN().compareTo(nextReadDLSN) >= 0);

                        // either there is data to read in current log segment or we are moving over a log segment that is
                        // still inprogress or was inprogress, we have check (or maybe close) this log segment.
                        final boolean checkOrCloseLedger = hasDataToRead ||
                            // next read position move over a log segment, if l is still inprogress or it was inprogress
                            ((l.isInProgress() || (null != currentMetadata && currentMetadata.isInProgress())) &&
                                    l.getLogSegmentSequenceNumber() == nextReadAheadPosition.getLogSegmentSequenceNumber());

                        // If we are positioning on a partially truncated log segment then the truncation point should
                        // be before the nextReadPosition
                        if (l.isPartiallyTruncated() &&
                            !isInitialPositioning &&
                            (l.getMinActiveDLSN().compareTo(nextReadDLSN) > 0)) {
                            if (conf.getAlertWhenPositioningOnTruncated()) {
                                alertStatsLogger.raise("Trying to position reader on {} when {} is marked partially truncated",
                                    nextReadAheadPosition, l);
                            }

                            if (!conf.getIgnoreTruncationStatus()) {
                                LOG.error("{}: Trying to position reader on {} when {} is marked partially truncated",
                                    new Object[]{ logMetadata.getFullyQualifiedName(), nextReadAheadPosition, l});
                                setReadingFromTruncated(tracker);
                                return;
                            }
                        }


                        if (LOG.isTraceEnabled()) {
                            LOG.trace("CheckLogSegment : newMetadata = {}, currentMetadata = {}, nextReadAheadPosition = {}",
                                      new Object[] { l, currentMetadata, nextReadAheadPosition});
                        }

                        if (checkOrCloseLedger) {
                            long startBKEntry = 0;
                            if (l.isPartiallyTruncated() && !conf.getIgnoreTruncationStatus()) {
                                startBKEntry = l.getMinActiveDLSN().getEntryId();
                                readAheadCache.setMinActiveDLSN(l.getMinActiveDLSN());
                            }

                            if(l.getLogSegmentSequenceNumber() == nextReadAheadPosition.getLogSegmentSequenceNumber()) {
                                startBKEntry = Math.max(startBKEntry, nextReadAheadPosition.getEntryId());
                                if (currentMetadata != null) {
                                    inProgressChanged = currentMetadata.isInProgress() && !l.isInProgress();
                                }
                            } else {
                                // We are positioning on a new ledger => reset the current ledger handle
                                LOG.trace("Positioning {} on a new ledger {}", fullyQualifiedName, l);

                                if (!closeCurrentLedgerHandle()) {
                                    return;
                                }
                            }

                            nextReadAheadPosition = new LedgerReadPosition(l.getLedgerId(), l.getLogSegmentSequenceNumber(), startBKEntry);
                            if (conf.getTraceReadAheadMetadataChanges()) {
                                LOG.info("Moved read position to {} for stream {} at {}.",
                                         new Object[] {nextReadAheadPosition, logMetadata.getFullyQualifiedName(), System.currentTimeMillis() });
                            }

                            if (l.isTruncated()) {
                                if (conf.getAlertWhenPositioningOnTruncated()) {
                                    alertStatsLogger.raise("Trying to position reader on {} when {} is marked truncated",
                                        nextReadAheadPosition, l);
                                }

                                if (!conf.getIgnoreTruncationStatus()) {
