/**
 * 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 org.apache.pulsar.broker.transaction.buffer.impl;

import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.collections4.map.LinkedMap;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.TransactionBuffer;
import org.apache.pulsar.broker.transaction.buffer.TransactionBufferReader;
import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
import org.apache.pulsar.broker.transaction.buffer.matadata.AbortTxnMetadata;
import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.RecoverTimeRecord;
import org.jctools.queues.MessagePassingQueue;
import org.jctools.queues.SpscArrayQueue;

/**
 * Transaction buffer based on normal persistent topic.
 */
@Slf4j
public class TopicTransactionBuffer extends TopicTransactionBufferState implements TransactionBuffer, TimerTask {

    private final PersistentTopic topic;

    private volatile PositionImpl maxReadPosition;

    /**
     * Ongoing transaction, map for remove txn stable position, linked for find max read position.
     */
    private final LinkedMap<TxnID, PositionImpl> ongoingTxns = new LinkedMap<>();

    /**
     * Aborts, map for jude message is aborted, linked for remove abort txn in memory when this
     * position have been deleted.
     */
    private final LinkedMap<TxnID, PositionImpl> aborts = new LinkedMap<>();

    private final CompletableFuture<SystemTopicClient.Writer<TransactionBufferSnapshot>> takeSnapshotWriter;

    // when add abort or change max read position, the count will +1. Take snapshot will set 0 into it.
    private final AtomicLong changeMaxReadPositionAndAddAbortTimes = new AtomicLong();

    private final LongAdder txnCommittedCounter = new LongAdder();

    private final LongAdder txnAbortedCounter = new LongAdder();

    private final Timer timer;

    private final int takeSnapshotIntervalNumber;

    private final int takeSnapshotIntervalTime;

    private volatile long lastSnapshotTimestamps;

    private final CompletableFuture<Void> transactionBufferFuture = new CompletableFuture<>();

    /**
     * The map is used to store the lowWaterMarks which key is TC ID and value is lowWaterMark of the TC.
     */
    private final ConcurrentHashMap<Long, Long> lowWaterMarks = new ConcurrentHashMap<>();

    public final RecoverTimeRecord recoverTime = new RecoverTimeRecord();

    private final Semaphore handleLowWaterMark = new Semaphore(1);

    public TopicTransactionBuffer(PersistentTopic topic) {
        super(State.None);
        this.topic = topic;
        this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar()
                .getTransactionBufferSnapshotService().createWriter(TopicName.get(topic.getName()));
        this.timer = topic.getBrokerService().getPulsar().getTransactionTimer();
        this.takeSnapshotIntervalNumber = topic.getBrokerService().getPulsar()
                .getConfiguration().getTransactionBufferSnapshotMaxTransactionCount();
        this.takeSnapshotIntervalTime = topic.getBrokerService().getPulsar()
                .getConfiguration().getTransactionBufferSnapshotMinTimeInMillis();
        this.maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
        this.recover();
    }

    private void recover() {
        recoverTime.setRecoverStartTime(System.currentTimeMillis());
        this.topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor(this)
                .execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() {
                    @Override
                    public void recoverComplete() {
                        synchronized (TopicTransactionBuffer.this) {
                            // sync maxReadPosition change to LAC when TopicTransaction buffer have not recover
                            // completely the normal message have been sent to broker and state is
                            // not Ready can't sync maxReadPosition when no ongoing transactions
                            if (ongoingTxns.isEmpty()) {
                                maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
                            }
                            if (!changeToReadyState()) {
                                log.error("[{}]Transaction buffer recover fail, current state: {}",
                                        topic.getName(), getState());
                                transactionBufferFuture.completeExceptionally
                                        (new BrokerServiceException.ServiceUnitNotReadyException(
                                                "Transaction buffer recover failed to change the status to Ready,"
                                                        + "current state is: " + getState()));
                            } else {
                                timer.newTimeout(TopicTransactionBuffer.this,
                                        takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
                                transactionBufferFuture.complete(null);
                                recoverTime.setRecoverEndTime(System.currentTimeMillis());
                            }
                        }
                    }

                    @Override
                    public void noNeedToRecover() {
                        synchronized (TopicTransactionBuffer.this) {
                            // sync maxReadPosition change to LAC when TopicTransaction buffer have not recover
                            // completely the normal message have been sent to broker and state is
                            // not NoSnapshot can't sync maxReadPosition
                            maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
                            if (!changeToNoSnapshotState()) {
                                log.error("[{}]Transaction buffer recover fail", topic.getName());
                            } else {
                                transactionBufferFuture.complete(null);
                                recoverTime.setRecoverEndTime(System.currentTimeMillis());
                            }
                        }
                    }

                    @Override
                    public void handleSnapshot(TransactionBufferSnapshot snapshot) {
                        maxReadPosition = PositionImpl.get(snapshot.getMaxReadPositionLedgerId(),
                                snapshot.getMaxReadPositionEntryId());
                        if (snapshot.getAborts() != null) {
                            snapshot.getAborts().forEach(abortTxnMetadata ->
                                    aborts.put(new TxnID(abortTxnMetadata.getTxnIdMostBits(),
                                                    abortTxnMetadata.getTxnIdLeastBits()),
                                            PositionImpl.get(abortTxnMetadata.getLedgerId(),
                                                    abortTxnMetadata.getEntryId())));
                        }
                    }

                    @Override
                    public void handleTxnEntry(Entry entry) {
                        ByteBuf metadataAndPayload = entry.getDataBuffer();

                        MessageMetadata msgMetadata = Commands.peekMessageMetadata(metadataAndPayload,
                                TopicTransactionBufferRecover.SUBSCRIPTION_NAME, -1);
                        if (msgMetadata != null && msgMetadata.hasTxnidMostBits() && msgMetadata.hasTxnidLeastBits()) {
                            TxnID txnID = new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits());
                            PositionImpl position = PositionImpl.get(entry.getLedgerId(), entry.getEntryId());
                            if (Markers.isTxnMarker(msgMetadata)) {
                                if (Markers.isTxnAbortMarker(msgMetadata)) {
                                    aborts.put(txnID, position);
                                }
                                updateMaxReadPosition(txnID);
                            } else {
                                handleTransactionMessage(txnID, position);
                            }
                        }
                    }

                    @Override
                    public void recoverExceptionally(Throwable e) {

                        log.warn("Closing topic {} due to read transaction buffer snapshot while recovering the "
                                + "transaction buffer throw exception", topic.getName(), e);
                        // when create reader or writer fail throw PulsarClientException,
                        // should close this topic and then reinit this topic
                        if (e instanceof PulsarClientException) {
                            // if transaction buffer recover fail throw PulsarClientException,
                            // we need to change the PulsarClientException to ServiceUnitNotReadyException,
                            // the tc do op will retry
                            transactionBufferFuture.completeExceptionally
                                    (new BrokerServiceException.ServiceUnitNotReadyException(e.getMessage(), e));
                        } else {
                            transactionBufferFuture.completeExceptionally(e);
                        }
                        recoverTime.setRecoverEndTime(System.currentTimeMillis());
                        topic.close(true);
                    }
                }, this.topic, this, takeSnapshotWriter));
    }

    @Override
    public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
        return CompletableFuture.completedFuture(null);
    }

    @Override
    public CompletableFuture<Void> checkIfTBRecoverCompletely(boolean isTxnEnabled) {
        if (!isTxnEnabled) {
            return CompletableFuture.completedFuture(null);
        } else {
            CompletableFuture<Void> completableFuture = new CompletableFuture<>();
            transactionBufferFuture.thenRun(() -> {
                if (checkIfNoSnapshot()) {
                    takeSnapshot().thenRun(() -> {
                        if (changeToReadyStateFromNoSnapshot()) {
                            timer.newTimeout(TopicTransactionBuffer.this,
                                    takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
                        }
                        completableFuture.complete(null);
                    }).exceptionally(exception -> {
                        log.error("Topic {} failed to take snapshot", this.topic.getName());
                        completableFuture.completeExceptionally(exception);
                        return null;
                    });
                } else {
                    completableFuture.complete(null);
                }
            }).exceptionally(exception -> {
                log.error("Topic {}: TransactionBuffer recover failed", this.topic.getName(), exception.getCause());
                completableFuture.completeExceptionally(exception.getCause());
                return null;
            });
            return completableFuture;
        }
    }

    @Override
    public long getOngoingTxnCount() {
        return this.ongoingTxns.size();
    }

    @Override
    public long getAbortedTxnCount() {
        return this.txnAbortedCounter.sum();
    }

    @Override
    public long getCommittedTxnCount() {
        return this.txnCommittedCounter.sum();
    }

    @Override
    public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceId, ByteBuf buffer) {
        CompletableFuture<Position> completableFuture = new CompletableFuture<>();
        Long lowWaterMark = lowWaterMarks.get(txnId.getMostSigBits());
        if (lowWaterMark != null && lowWaterMark >= txnId.getLeastSigBits()) {
            completableFuture.completeExceptionally(new BrokerServiceException
                    .NotAllowedException("Transaction [" + txnId + "] has been ended. "
                    + "Please use a new transaction to send message."));
            return completableFuture;
        }
        topic.getManagedLedger().asyncAddEntry(buffer, new AsyncCallbacks.AddEntryCallback() {
            @Override
            public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                synchronized (TopicTransactionBuffer.this) {
                    handleTransactionMessage(txnId, position);
                }
                completableFuture.complete(position);
            }

            @Override
            public void addFailed(ManagedLedgerException exception, Object ctx) {
                log.error("Failed to append buffer to txn {}", txnId, exception);
                completableFuture.completeExceptionally(exception);
            }
        }, null);
        return completableFuture;
    }

    private void handleTransactionMessage(TxnID txnId, Position position) {
        if (!ongoingTxns.containsKey(txnId) && !aborts.containsKey(txnId)) {
            ongoingTxns.put(txnId, (PositionImpl) position);
            PositionImpl firstPosition = ongoingTxns.get(ongoingTxns.firstKey());
            //max read position is less than first ongoing transaction message position, so entryId -1
            maxReadPosition = PositionImpl.get(firstPosition.getLedgerId(), firstPosition.getEntryId() - 1);
        }
    }


    @Override
    public CompletableFuture<TransactionBufferReader> openTransactionBufferReader(TxnID txnID, long startSequenceId) {
        return null;
    }

    @Override
    public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
        if (log.isDebugEnabled()) {
            log.debug("Transaction {} commit on topic {}.", txnID.toString(), topic.getName());
        }
        CompletableFuture<Void> completableFuture = new CompletableFuture<>();
        //Wait TB recover completely.
        transactionBufferFuture.thenRun(() -> {
            ByteBuf commitMarker = Markers.newTxnCommitMarker(-1L, txnID.getMostSigBits(),
                    txnID.getLeastSigBits());
            try {
                topic.getManagedLedger().asyncAddEntry(commitMarker, new AsyncCallbacks.AddEntryCallback() {
                    @Override
                    public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                        synchronized (TopicTransactionBuffer.this) {
                            updateMaxReadPosition(txnID);
                            handleLowWaterMark(txnID, lowWaterMark);
                            clearAbortedTransactions();
                            takeSnapshotByChangeTimes();
                        }
                        txnCommittedCounter.increment();
                        completableFuture.complete(null);
                    }

                    @Override
                    public void addFailed(ManagedLedgerException exception, Object ctx) {
                        log.error("Failed to commit for txn {}", txnID, exception);
                        checkAppendMarkerException(exception);
                        completableFuture.completeExceptionally(new PersistenceException(exception));
                    }
                }, null);
            } finally {
                commitMarker.release();
            }
        }).exceptionally(exception -> {
            log.error("Transaction {} commit on topic {}.", txnID.toString(), topic.getName(), exception.getCause());
            completableFuture.completeExceptionally(exception.getCause());
            return null;
        });
        return completableFuture;
    }

    @Override
    public CompletableFuture<Void> abortTxn(TxnID txnID, long lowWaterMark) {
        if (log.isDebugEnabled()) {
            log.debug("Transaction {} abort on topic {}.", txnID.toString(), topic.getName());
        }
        CompletableFuture<Void> completableFuture = new CompletableFuture<>();
        //Wait TB recover completely.
        transactionBufferFuture.thenRun(() -> {
            //no message sent, need not to add abort mark by txn timeout.
            if (!checkIfReady()) {
                completableFuture.complete(null);
                return;
            }
            ByteBuf abortMarker = Markers.newTxnAbortMarker(-1L, txnID.getMostSigBits(), txnID.getLeastSigBits());
            try {
                topic.getManagedLedger().asyncAddEntry(abortMarker, new AsyncCallbacks.AddEntryCallback() {
                    @Override
                    public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                        synchronized (TopicTransactionBuffer.this) {
                            aborts.put(txnID, (PositionImpl) position);
                            updateMaxReadPosition(txnID);
                            changeMaxReadPositionAndAddAbortTimes.getAndIncrement();
                            clearAbortedTransactions();
                            takeSnapshotByChangeTimes();
                        }
                        txnAbortedCounter.increment();
                        completableFuture.complete(null);
                        handleLowWaterMark(txnID, lowWaterMark);
                    }

                    @Override
                    public void addFailed(ManagedLedgerException exception, Object ctx) {
                        log.error("Failed to abort for txn {}", txnID, exception);
                        checkAppendMarkerException(exception);
                        completableFuture.completeExceptionally(new PersistenceException(exception));
                    }
                }, null);
            } finally {
                abortMarker.release();
            }
        }).exceptionally(exception -> {
            log.error("Transaction {} abort on topic {}.", txnID.toString(), topic.getName(), exception.getCause());
            completableFuture.completeExceptionally(exception.getCause());
            return null;
        });
        return completableFuture;
    }

    private void checkAppendMarkerException(ManagedLedgerException exception) {
        if (exception instanceof ManagedLedgerException.ManagedLedgerAlreadyClosedException) {
            topic.getManagedLedger().readyToCreateNewLedger();
        }
    }

    private void handleLowWaterMark(TxnID txnID, long lowWaterMark) {
        lowWaterMarks.compute(txnID.getMostSigBits(), (tcId, oldLowWaterMark) -> {
            if (oldLowWaterMark == null || oldLowWaterMark < lowWaterMark) {
                return lowWaterMark;
            } else {
                return oldLowWaterMark;
            }
        });
        if (handleLowWaterMark.tryAcquire()) {
            if (!ongoingTxns.isEmpty()) {
                TxnID firstTxn = ongoingTxns.firstKey();
                long tCId = firstTxn.getMostSigBits();
                Long lowWaterMarkOfFirstTxnId = lowWaterMarks.get(tCId);
                if (lowWaterMarkOfFirstTxnId != null && firstTxn.getLeastSigBits() <= lowWaterMarkOfFirstTxnId) {
                    abortTxn(firstTxn, lowWaterMarkOfFirstTxnId)
                            .thenRun(() -> {
                                log.warn("Successes to abort low water mark for txn [{}], topic [{}],"
                                        + " lowWaterMark [{}]", firstTxn, topic.getName(), lowWaterMarkOfFirstTxnId);
                                handleLowWaterMark.release();
                            })
                            .exceptionally(ex -> {
                                log.warn("Failed to abort low water mark for txn {}, topic [{}], "
                                        + "lowWaterMark [{}], ", firstTxn, topic.getName(), lowWaterMarkOfFirstTxnId,
                                        ex);
                                handleLowWaterMark.release();
                                return null;
                            });
                    return;
                }
            }
            handleLowWaterMark.release();
        }
    }

    private void takeSnapshotByChangeTimes() {
        if (changeMaxReadPositionAndAddAbortTimes.get() >= takeSnapshotIntervalNumber) {
            takeSnapshot();
        }
    }

    private void takeSnapshotByTimeout() {
        if (changeMaxReadPositionAndAddAbortTimes.get() > 0) {
            takeSnapshot();
        }
        this.timer.newTimeout(TopicTransactionBuffer.this,
                takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
    }

    private CompletableFuture<Void> takeSnapshot() {
        changeMaxReadPositionAndAddAbortTimes.set(0);
        return takeSnapshotWriter.thenCompose(writer -> {
            TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot();
            synchronized (TopicTransactionBuffer.this) {
                snapshot.setTopicName(topic.getName());
                snapshot.setMaxReadPositionLedgerId(maxReadPosition.getLedgerId());
                snapshot.setMaxReadPositionEntryId(maxReadPosition.getEntryId());
                List<AbortTxnMetadata> list = new ArrayList<>();
                aborts.forEach((k, v) -> {
                    AbortTxnMetadata abortTxnMetadata = new AbortTxnMetadata();
                    abortTxnMetadata.setTxnIdMostBits(k.getMostSigBits());
                    abortTxnMetadata.setTxnIdLeastBits(k.getLeastSigBits());
                    abortTxnMetadata.setLedgerId(v.getLedgerId());
                    abortTxnMetadata.setEntryId(v.getEntryId());
                    list.add(abortTxnMetadata);
                });
                snapshot.setAborts(list);
            }
            return writer.writeAsync(snapshot).thenAccept(messageId-> {
                this.lastSnapshotTimestamps = System.currentTimeMillis();
                if (log.isDebugEnabled()) {
                    log.debug("[{}]Transaction buffer take snapshot success! "
                            + "messageId : {}", topic.getName(), messageId);
                }
            }).exceptionally(e -> {
                log.warn("[{}]Transaction buffer take snapshot fail! ", topic.getName(), e);
                return null;
            });
        });
    }
    private void clearAbortedTransactions() {
        while (!aborts.isEmpty() && !((ManagedLedgerImpl) topic.getManagedLedger())
                .ledgerExists(aborts.get(aborts.firstKey()).getLedgerId())) {
            if (log.isDebugEnabled()) {
                aborts.firstKey();
                log.debug("[{}] Topic transaction buffer clear aborted transaction, TxnId : {}, Position : {}",
                        topic.getName(), aborts.firstKey(), aborts.get(aborts.firstKey()));
            }
            aborts.remove(aborts.firstKey());
        }
    }
    void updateMaxReadPosition(TxnID txnID) {
        PositionImpl preMaxReadPosition = this.maxReadPosition;
        ongoingTxns.remove(txnID);
        if (!ongoingTxns.isEmpty()) {
            PositionImpl position = ongoingTxns.get(ongoingTxns.firstKey());
            //max read position is less than first ongoing transaction message position, so entryId -1
            maxReadPosition = PositionImpl.get(position.getLedgerId(), position.getEntryId() - 1);
        } else {
            maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
        }
        if (preMaxReadPosition.compareTo(this.maxReadPosition) != 0) {
            this.changeMaxReadPositionAndAddAbortTimes.getAndIncrement();
        }
    }

    @Override
    public CompletableFuture<Void> purgeTxns(List<Long> dataLedgers) {
        return null;
    }

    @Override
    public CompletableFuture<Void> clearSnapshot() {
        return this.takeSnapshotWriter.thenCompose(writer -> {
            TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot();
            snapshot.setTopicName(topic.getName());
            return writer.deleteAsync(snapshot);
        }).thenCompose(__ -> CompletableFuture.completedFuture(null));
    }

    @Override
    public CompletableFuture<Void> closeAsync() {
        changeToCloseState();
        return this.takeSnapshotWriter.thenCompose(SystemTopicClient.Writer::closeAsync);
    }

    @Override
    public boolean isTxnAborted(TxnID txnID) {
        return aborts.containsKey(txnID);
    }

    @Override
    public void syncMaxReadPositionForNormalPublish(PositionImpl position) {
        // when ongoing transaction is empty, proved that lastAddConfirm is can read max position, because callback
        // thread is the same tread, in this time the lastAddConfirm don't content transaction message.
        synchronized (TopicTransactionBuffer.this) {
            if (checkIfNoSnapshot()) {
                maxReadPosition = position;
            } else if (checkIfReady()) {
                if (ongoingTxns.isEmpty()) {
                    maxReadPosition = position;
                    changeMaxReadPositionAndAddAbortTimes.incrementAndGet();
                }
            }
        }
    }

    @Override
    public PositionImpl getMaxReadPosition() {
        if (checkIfReady() || checkIfNoSnapshot()) {
            return this.maxReadPosition;
        } else {
            return PositionImpl.EARLIEST;
        }
    }

    @Override
    public TransactionInBufferStats getTransactionInBufferStats(TxnID txnID) {
        TransactionInBufferStats transactionInBufferStats = new TransactionInBufferStats();
        transactionInBufferStats.aborted = isTxnAborted(txnID);
        if (ongoingTxns.containsKey(txnID)) {
            transactionInBufferStats.startPosition = ongoingTxns.get(txnID).toString();
        }
        return transactionInBufferStats;
    }

    @Override
    public TransactionBufferStats getStats(boolean lowWaterMarks) {
        TransactionBufferStats transactionBufferStats = new TransactionBufferStats();
        transactionBufferStats.lastSnapshotTimestamps = this.lastSnapshotTimestamps;
        transactionBufferStats.state = this.getState().name();
        transactionBufferStats.maxReadPosition = this.maxReadPosition.toString();
        if (lowWaterMarks) {
            transactionBufferStats.lowWaterMarks = this.lowWaterMarks;
        }
        transactionBufferStats.ongoingTxnSize = ongoingTxns.size();

        transactionBufferStats.recoverStartTime = recoverTime.getRecoverStartTime();
        transactionBufferStats.recoverEndTime = recoverTime.getRecoverEndTime();
        return transactionBufferStats;
    }

    @Override
    public void run(Timeout timeout) {
        if (checkIfReady()) {
            takeSnapshotByTimeout();
        }
    }

    // we store the maxReadPosition from snapshot then open the non-durable cursor by this topic's manageLedger.
    // the non-durable cursor will read to lastConfirmedEntry.
    @VisibleForTesting
    public static class TopicTransactionBufferRecover implements Runnable {

        private final PersistentTopic topic;

        private final TopicTransactionBufferRecoverCallBack callBack;

        private Position startReadCursorPosition = PositionImpl.EARLIEST;

        private final SpscArrayQueue<Entry> entryQueue;

        private final AtomicLong exceptionNumber = new AtomicLong();

        public static final String SUBSCRIPTION_NAME = "transaction-buffer-sub";

        private final TopicTransactionBuffer topicTransactionBuffer;

        private final CompletableFuture<SystemTopicClient.Writer<TransactionBufferSnapshot>> takeSnapshotWriter;

        private TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack callBack, PersistentTopic topic,
                                              TopicTransactionBuffer transactionBuffer, CompletableFuture<
                SystemTopicClient.Writer<TransactionBufferSnapshot>> takeSnapshotWriter) {
            this.topic = topic;
            this.callBack = callBack;
            this.entryQueue = new SpscArrayQueue<>(2000);
            this.topicTransactionBuffer = transactionBuffer;
            this.takeSnapshotWriter = takeSnapshotWriter;
        }

        @SneakyThrows
        @Override
        public void run() {
            this.takeSnapshotWriter.thenRunAsync(() -> {
                if (!this.topicTransactionBuffer.changeToInitializingState()) {
                    log.warn("TransactionBuffer {} of topic {} can not change state to Initializing",
                            this, topic.getName());
                    return;
                }
                topic.getBrokerService().getPulsar().getTransactionBufferSnapshotService()
                        .createReader(TopicName.get(topic.getName())).thenAcceptAsync(reader -> {
                            try {
                                boolean hasSnapshot = false;
                                while (reader.hasMoreEvents()) {
                                    Message<TransactionBufferSnapshot> message = reader.readNext();
                                    if (topic.getName().equals(message.getKey())) {
                                        TransactionBufferSnapshot transactionBufferSnapshot = message.getValue();
                                        if (transactionBufferSnapshot != null) {
                                            hasSnapshot = true;
                                            callBack.handleSnapshot(transactionBufferSnapshot);
                                            this.startReadCursorPosition = PositionImpl.get(
                                                    transactionBufferSnapshot.getMaxReadPositionLedgerId(),
                                                    transactionBufferSnapshot.getMaxReadPositionEntryId());
                                        }
                                    }
                                }
                                if (!hasSnapshot) {
                                    closeReader(reader);
                                    callBack.noNeedToRecover();
                                    return;
                                }
                            } catch (Exception ex) {
                                log.error("[{}] Transaction buffer recover fail when read "
                                        + "transactionBufferSnapshot!", topic.getName(), ex);
                                callBack.recoverExceptionally(ex);
                                closeReader(reader);
                                return;
                            }
                            closeReader(reader);

                            ManagedCursor managedCursor;
                            try {
                                managedCursor = topic.getManagedLedger()
                                        .newNonDurableCursor(this.startReadCursorPosition, SUBSCRIPTION_NAME);
                            } catch (ManagedLedgerException e) {
                                callBack.recoverExceptionally(e);
                                log.error("[{}]Transaction buffer recover fail when open cursor!", topic.getName(), e);
                                return;
                            }
                            PositionImpl lastConfirmedEntry =
                                    (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
                            PositionImpl currentLoadPosition = (PositionImpl) this.startReadCursorPosition;
                            FillEntryQueueCallback fillEntryQueueCallback = new FillEntryQueueCallback(entryQueue,
                                    managedCursor, TopicTransactionBufferRecover.this);
                            if (lastConfirmedEntry.getEntryId() != -1) {
                                while (lastConfirmedEntry.compareTo(currentLoadPosition) > 0
                                        && fillEntryQueueCallback.fillQueue()) {
                                    Entry entry = entryQueue.poll();
                                    if (entry != null) {
                                        try {
                                            currentLoadPosition = PositionImpl.get(entry.getLedgerId(),
                                                    entry.getEntryId());
                                            callBack.handleTxnEntry(entry);
                                        } finally {
                                            entry.release();
                                        }
                                    } else {
                                        try {
                                            Thread.sleep(1);
                                        } catch (InterruptedException e) {
                                            //no-op
                                        }
                                    }
                                }
                            }

                            closeCursor(SUBSCRIPTION_NAME);
                            callBack.recoverComplete();
                        }, topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
                                .getExecutor(this)).exceptionally(e -> {
                            callBack.recoverExceptionally(e.getCause());
                            log.error("[{}]Transaction buffer new snapshot reader fail!", topic.getName(), e);
                            return null;
                        });
            }, topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
                    .getExecutor(this)).exceptionally(e -> {
                callBack.recoverExceptionally(e.getCause());
                log.error("[{}]Transaction buffer create snapshot writer fail!",
                        topic.getName(), e);
                return null;
            });
        }

        private void closeCursor(String subscriptionName) {
            topic.getManagedLedger().asyncDeleteCursor(Codec.encode(subscriptionName),
                    new AsyncCallbacks.DeleteCursorCallback() {
                @Override
                public void deleteCursorComplete(Object ctx) {
                    log.info("[{}]Transaction buffer snapshot recover cursor close complete.", topic.getName());
                }

                @Override
                public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
                    log.error("[{}]Transaction buffer snapshot recover cursor close fail.", topic.getName());
                }

            }, null);
        }

        private void callBackException(ManagedLedgerException e) {
            log.error("Transaction buffer recover fail when recover transaction entry!", e);
            this.exceptionNumber.getAndIncrement();
        }

        private void closeReader(SystemTopicClient.Reader<TransactionBufferSnapshot> reader) {
            reader.closeAsync().exceptionally(e -> {
                log.error("[{}]Transaction buffer reader close error!", topic.getName(), e);
                return null;
            });
        }
    }

    static class FillEntryQueueCallback implements AsyncCallbacks.ReadEntriesCallback {

        private final AtomicLong outstandingReadsRequests = new AtomicLong(0);

        private final SpscArrayQueue<Entry> entryQueue;

        private final ManagedCursor cursor;

        private final TopicTransactionBufferRecover recover;

        private volatile boolean isReadable = true;

        private static final int NUMBER_OF_PER_READ_ENTRY = 100;

        private FillEntryQueueCallback(SpscArrayQueue<Entry> entryQueue, ManagedCursor cursor,
                                       TopicTransactionBufferRecover recover) {
            this.entryQueue = entryQueue;
            this.cursor = cursor;
            this.recover = recover;
        }
        boolean fillQueue() {
            if (entryQueue.size() + NUMBER_OF_PER_READ_ENTRY < entryQueue.capacity()
                    && outstandingReadsRequests.get() == 0) {
                if (cursor.hasMoreEntries()) {
                    outstandingReadsRequests.incrementAndGet();
                    cursor.asyncReadEntries(NUMBER_OF_PER_READ_ENTRY,
                            this, System.nanoTime(), PositionImpl.LATEST);
                } else {
                    if (entryQueue.size() == 0) {
                        isReadable = false;
                    }
                }
            }
            return isReadable;
        }

        @Override
        public void readEntriesComplete(List<Entry> entries, Object ctx) {
            entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
                private int i = 0;
                @Override
                public Entry get() {
                    Entry entry = entries.get(i);
                    i++;
                    return entry;
                }
            }, entries.size());

            outstandingReadsRequests.decrementAndGet();
        }

        @Override
        public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
            if (recover.topic.getManagedLedger().getConfig().isAutoSkipNonRecoverableData()
                    && exception instanceof ManagedLedgerException.NonRecoverableLedgerException
                    || exception instanceof ManagedLedgerException.ManagedLedgerFencedException
                    || exception instanceof ManagedLedgerException.CursorAlreadyClosedException) {
                isReadable = false;
            } else {
                outstandingReadsRequests.decrementAndGet();
            }
            recover.callBackException(exception);
        }
    }
}