/**
 * 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.bookkeeper.mledger.impl;

import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import com.google.common.collect.Sets;
import java.lang.reflect.Field;
import java.nio.charset.Charset;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
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;
import java.util.stream.Collectors;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import lombok.Cleanup;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedCursor.IndividualDeletedEntries;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.common.util.collections.LongPairRangeSet;
import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.Stat;
import org.apache.pulsar.common.api.proto.IntRange;
import org.awaitility.Awaitility;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

public class ManagedCursorTest extends MockedBookKeeperTestCase {

    private static final Charset Encoding = Charsets.UTF_8;

    @DataProvider(name = "useOpenRangeSet")
    public static Object[][] useOpenRangeSet() {
        return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
    }


    @Test(timeOut = 20000)
    void readFromEmptyLedger() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursor c1 = ledger.openCursor("c1");
        List<Entry> entries = c1.readEntries(10);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());

        ledger.addEntry("test".getBytes(Encoding));
        entries = c1.readEntries(10);
        assertEquals(entries.size(), 1);
        entries.forEach(e -> e.release());

        entries = c1.readEntries(10);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());

        // Test string representation
        assertEquals(c1.toString(), "ManagedCursorImpl{ledger=my_test_ledger, name=c1, ackPos=3:-1, readPos=3:1}");
    }

    @Test(timeOut = 20000)
    void readTwice() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");
        ManagedCursor c2 = ledger.openCursor("c2");

        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(2);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());

        entries = c1.readEntries(2);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());

        entries = c2.readEntries(2);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());

        entries = c2.readEntries(2);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void readWithCacheDisabled() throws Exception {
        ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig();
        config.setMaxCacheSize(0);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config);
        ManagedLedger ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");
        ManagedCursor c2 = ledger.openCursor("c2");

        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(2);
        assertEquals(entries.size(), 2);
        assertEquals(new String(entries.get(0).getData(), Encoding), "entry-1");
        assertEquals(new String(entries.get(1).getData(), Encoding), "entry-2");
        entries.forEach(e -> e.release());

        entries = c1.readEntries(2);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());

        entries = c2.readEntries(2);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());

        entries = c2.readEntries(2);
        assertEquals(entries.size(), 0);
        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void getEntryDataTwice() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursor c1 = ledger.openCursor("c1");

        ledger.addEntry("entry-1".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(2);
        assertEquals(entries.size(), 1);

        Entry entry = entries.get(0);
        assertEquals(entry.getLength(), "entry-1".length());
        byte[] data1 = entry.getData();
        byte[] data2 = entry.getData();
        assertEquals(data1, data2);
        entry.release();
    }

    @Test(timeOut = 20000)
    void readFromClosedLedger() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");

        ledger.close();

        try {
            c1.readEntries(2);
            fail("ledger is closed, should fail");
        } catch (ManagedLedgerException e) {
            // ok
        }
    }

    @Test(timeOut = 20000)
    void testNumberOfEntries() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));

        ManagedCursor c1 = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ManagedCursor c3 = ledger.openCursor("c3");
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ManagedCursor c4 = ledger.openCursor("c4");
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ManagedCursor c5 = ledger.openCursor("c5");

        assertEquals(c1.getNumberOfEntries(), 4);
        assertTrue(c1.hasMoreEntries());

        assertEquals(c2.getNumberOfEntries(), 3);
        assertTrue(c2.hasMoreEntries());

        assertEquals(c3.getNumberOfEntries(), 2);
        assertTrue(c3.hasMoreEntries());

        assertEquals(c4.getNumberOfEntries(), 1);
        assertTrue(c4.hasMoreEntries());

        assertEquals(c5.getNumberOfEntries(), 0);
        assertFalse(c5.hasMoreEntries());

        List<Entry> entries = c1.readEntries(2);
        assertEquals(entries.size(), 2);
        c1.markDelete(entries.get(1).getPosition());
        assertEquals(c1.getNumberOfEntries(), 2);
        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void testNumberOfEntriesInBacklog() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));

        ManagedCursor c1 = ledger.openCursor("c1");
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ManagedCursor c3 = ledger.openCursor("c3");
        Position p3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ManagedCursor c4 = ledger.openCursor("c4");
        Position p4 = ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ManagedCursor c5 = ledger.openCursor("c5");

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 4);
        assertEquals(c2.getNumberOfEntriesInBacklog(false), 3);
        assertEquals(c3.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c4.getNumberOfEntriesInBacklog(false), 1);
        assertEquals(c5.getNumberOfEntriesInBacklog(false), 0);

        List<Entry> entries = c1.readEntries(2);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());

        assertEquals(c1.getNumberOfEntries(), 2);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 4);

        c1.markDelete(p1);
        assertEquals(c1.getNumberOfEntries(), 2);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);

        c1.delete(p3);

        assertEquals(c1.getNumberOfEntries(), 1);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);

        c1.markDelete(p4);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
    }

    @Test(timeOut = 20000)
    void testNumberOfEntriesInBacklogWithFallback() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));

        ManagedCursor c1 = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ManagedCursor c3 = ledger.openCursor("c3");
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ManagedCursor c4 = ledger.openCursor("c4");
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ManagedCursor c5 = ledger.openCursor("c5");

        Field field = ManagedCursorImpl.class.getDeclaredField("messagesConsumedCounter");
        field.setAccessible(true);
        long counter = ((ManagedLedgerImpl) ledger).getEntriesAddedCounter() + 1;
        field.setLong(c1, counter);
        field.setLong(c2, counter);
        field.setLong(c3, counter);
        field.setLong(c4, counter);
        field.setLong(c5, counter);

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 4);
        assertEquals(c2.getNumberOfEntriesInBacklog(false), 3);
        assertEquals(c3.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c4.getNumberOfEntriesInBacklog(false), 1);
        assertEquals(c5.getNumberOfEntriesInBacklog(false), 0);
    }

    @Test(timeOut = 20000)
    void testNumberOfEntriesWithReopen() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ManagedCursor c3 = ledger.openCursor("c3");

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        c1 = ledger.openCursor("c1");
        c2 = ledger.openCursor("c2");
        c3 = ledger.openCursor("c3");

        assertEquals(c1.getNumberOfEntries(), 2);
        assertTrue(c1.hasMoreEntries());

        assertEquals(c2.getNumberOfEntries(), 1);
        assertTrue(c2.hasMoreEntries());

        assertEquals(c3.getNumberOfEntries(), 0);
        assertFalse(c3.hasMoreEntries());
    }

    @Test(timeOut = 20000)
    void asyncReadWithoutErrors() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");

        ledger.addEntry("dummy-entry-1".getBytes(Encoding));

        final CountDownLatch counter = new CountDownLatch(1);

        cursor.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                assertNull(ctx);
                assertEquals(entries.size(), 1);
                entries.forEach(e -> e.release());
                counter.countDown();
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                fail(exception.getMessage());
            }

        }, null, PositionImpl.latest);

        counter.await();
    }

    @Test(timeOut = 20000)
    void asyncReadWithErrors() throws Exception {
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");

        ledger.addEntry("dummy-entry-1".getBytes(Encoding));

        final CountDownLatch counter = new CountDownLatch(1);

        stopBookKeeper();

        cursor.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                entries.forEach(e -> e.release());
                counter.countDown();
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                fail("async-call should not have failed");
            }

        }, null, PositionImpl.latest);

        counter.await();

        cursor.rewind();

        // Clear the cache to force reading from BK
        ledger.entryCache.clear();

        final CountDownLatch counter2 = new CountDownLatch(1);

        cursor.asyncReadEntries(100, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                fail("async-call should have failed");
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                counter2.countDown();
            }

        }, null, PositionImpl.latest);

        counter2.await();
    }

    @Test(timeOut = 20000, expectedExceptions = IllegalArgumentException.class)
    void asyncReadWithInvalidParameter() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");

        ledger.addEntry("dummy-entry-1".getBytes(Encoding));

        final CountDownLatch counter = new CountDownLatch(1);

        stopBookKeeper();

        cursor.asyncReadEntries(0, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                fail("async-call should have failed");
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                counter.countDown();
            }

        }, null, PositionImpl.latest);

        counter.await();
    }

    @Test(timeOut = 20000)
    void testAsyncReadWithMaxSizeByte() throws Exception {
        ManagedLedger ledger = factory.open("testAsyncReadWithMaxSizeByte");
        ManagedCursor cursor = ledger.openCursor("c1");

        for (int i = 0; i < 100; i++) {
            ledger.addEntry(new byte[1024]);
        }

        // First time, since we don't have info, we'll get 1 single entry
        readAndCheck(cursor, 10, 3 * 1024, 1);
        // We should only return 3 entries, based on the max size
        readAndCheck(cursor, 20, 3 * 1024, 3);
        // If maxSize is < avg, we should get 1 entry
        readAndCheck(cursor, 10, 500, 1);
    }

    private void readAndCheck(ManagedCursor cursor, int numEntriesToRead,
                              long maxSizeBytes, int expectedNumRead) throws InterruptedException {
        CountDownLatch counter = new CountDownLatch(1);
        cursor.asyncReadEntries(numEntriesToRead, maxSizeBytes, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                Assert.assertEquals(entries.size(), expectedNumRead);
                entries.forEach(e -> e.release());
                counter.countDown();
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                fail(exception.getMessage());
            }
        }, null, null);
        counter.await();
    }

    @Test(timeOut = 20000)
    void markDeleteWithErrors() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        List<Entry> entries = cursor.readEntries(100);

        stopBookKeeper();
        assertEquals(entries.size(), 1);

        // Mark-delete should succeed if BK is down
        cursor.markDelete(entries.get(0).getPosition());

        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void markDeleteWithZKErrors() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        List<Entry> entries = cursor.readEntries(100);

        assertEquals(entries.size(), 1);

        stopBookKeeper();
        metadataStore.setAlwaysFail(new MetadataStoreException("error"));

        try {
            cursor.markDelete(entries.get(0).getPosition());
            fail("Should have failed");
        } catch (Exception e) {
            // Expected
        }

        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void markDeleteAcrossLedgers() throws Exception {
        ManagedLedger ml1 = factory.open("my_test_ledger");
        ManagedCursor mc1 = ml1.openCursor("c1");

        // open ledger id 3 for ml1
        // markDeletePosition for mc1 is 3:-1
        // readPosition is 3:0

        ml1.close();
        mc1.close();

        // force removal of this ledger from the cache
        factory.close(ml1);

        ManagedLedger ml2 = factory.open("my_test_ledger");
        ManagedCursor mc2 = ml2.openCursor("c1");

        // open ledger id 5 for ml2
        // this entry is written at 5:0
        Position pos = ml2.addEntry("dummy-entry-1".getBytes(Encoding));

        List<Entry> entries = mc2.readEntries(1);
        assertEquals(entries.size(), 1);
        assertEquals(new String(entries.get(0).getData(), Encoding), "dummy-entry-1");
        entries.forEach(e -> e.release());

        mc2.delete(pos);

        // verify if the markDeletePosition moves from 3:-1 to 5:0
        assertEquals(mc2.getMarkDeletedPosition(), pos);
        assertEquals(mc2.getMarkDeletedPosition().getNext(), mc2.getReadPosition());
    }

    @Test(timeOut = 20000)
    void testResetCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_move_cursor_ledger",
                new ManagedLedgerConfig().setMaxEntriesPerLedger(10));
        ManagedCursor cursor = ledger.openCursor("trc1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        final AtomicBoolean moveStatus = new AtomicBoolean(false);
        PositionImpl resetPosition = new PositionImpl(lastPosition.getLedgerId(), lastPosition.getEntryId() - 2);
        try {
            cursor.resetCursor(resetPosition);
            moveStatus.set(true);
        } catch (Exception e) {
            log.warn("error in reset cursor", e.getCause());
        }

        assertTrue(moveStatus.get());
        assertEquals(resetPosition, cursor.getReadPosition());
        cursor.close();
        ledger.close();
    }

    @Test(timeOut = 20000)
    void testResetCursor1() throws Exception {
        ManagedLedger ledger = factory.open("my_test_move_cursor_ledger",
            new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
        ManagedCursor cursor = ledger.openCursor("trc1");
        PositionImpl actualEarliest = (PositionImpl) ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl lastInPrev = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        PositionImpl firstInNext = (PositionImpl) ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));
        ledger.addEntry("dummy-entry-7".getBytes(Encoding));
        ledger.addEntry("dummy-entry-8".getBytes(Encoding));
        ledger.addEntry("dummy-entry-9".getBytes(Encoding));
        PositionImpl last = (PositionImpl) ledger.addEntry("dummy-entry-10".getBytes(Encoding));

        final AtomicBoolean moveStatus = new AtomicBoolean(false);

        // reset to earliest
        PositionImpl earliest = PositionImpl.earliest;
        try {
            cursor.resetCursor(earliest);
            moveStatus.set(true);
        } catch (Exception e) {
            log.warn("error in reset cursor", e.getCause());
        }
        assertTrue(moveStatus.get());
        PositionImpl earliestPos = new PositionImpl(actualEarliest.getLedgerId(), -1);
        assertEquals(earliestPos, cursor.getReadPosition());
        moveStatus.set(false);

        // reset to one after last entry in a ledger should point to the first entry in the next ledger
        PositionImpl resetPosition = new PositionImpl(lastInPrev.getLedgerId(), lastInPrev.getEntryId() + 1);
        try {
            cursor.resetCursor(resetPosition);
            moveStatus.set(true);
        } catch (Exception e) {
            log.warn("error in reset cursor", e.getCause());
        }
        assertTrue(moveStatus.get());
        assertEquals(firstInNext, cursor.getReadPosition());
        moveStatus.set(false);

        // reset to a non exist larger ledger should point to the first non-exist entry in the last ledger
        PositionImpl latest = new PositionImpl(last.getLedgerId() + 2, 0);
        try {
            cursor.resetCursor(latest);
            moveStatus.set(true);
        } catch (Exception e) {
            log.warn("error in reset cursor", e.getCause());
        }
        assertTrue(moveStatus.get());
        PositionImpl lastPos = new PositionImpl(last.getLedgerId(), last.getEntryId() + 1);
        assertEquals(lastPos, cursor.getReadPosition());
        moveStatus.set(false);

        // reset to latest should point to the first non-exist entry in the last ledger
        PositionImpl anotherLast = PositionImpl.latest;
        try {
            cursor.resetCursor(anotherLast);
            moveStatus.set(true);
        } catch (Exception e) {
            log.warn("error in reset cursor", e.getCause());
        }
        assertTrue(moveStatus.get());
        assertEquals(lastPos, cursor.getReadPosition());

        cursor.close();
        ledger.close();
    }

    @Test(timeOut = 20000)
    void testasyncResetCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_move_cursor_ledger",
                new ManagedLedgerConfig().setMaxEntriesPerLedger(10));
        ManagedCursor cursor = ledger.openCursor("tarc1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        final AtomicBoolean moveStatus = new AtomicBoolean(false);
        CountDownLatch countDownLatch = new CountDownLatch(1);
        PositionImpl resetPosition = new PositionImpl(lastPosition.getLedgerId(), lastPosition.getEntryId() - 2);

        cursor.asyncResetCursor(resetPosition, false, new AsyncCallbacks.ResetCursorCallback() {
            @Override
            public void resetComplete(Object ctx) {
                moveStatus.set(true);
                countDownLatch.countDown();
            }

            @Override
            public void resetFailed(ManagedLedgerException exception, Object ctx) {
                moveStatus.set(false);
                countDownLatch.countDown();
            }
        });
        countDownLatch.await();
        assertTrue(moveStatus.get());
        assertEquals(resetPosition, cursor.getReadPosition());
        cursor.close();
        ledger.close();
    }

    @Test(timeOut = 20000)
    void testConcurrentResetCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_concurrent_move_ledger");

        final int Messages = 100;
        final int Consumers = 5;

        List<Future<AtomicBoolean>> futures = Lists.newArrayList();
        @Cleanup("shutdownNow")
        ExecutorService executor = Executors.newCachedThreadPool();
        final CyclicBarrier barrier = new CyclicBarrier(Consumers + 1);

        for (int i = 0; i < Messages; i++) {
            ledger.addEntry("test".getBytes());
        }
        final PositionImpl lastPosition = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));

        for (int i = 0; i < Consumers; i++) {
            final ManagedCursor cursor = ledger.openCursor("tcrc" + i);
            final int idx = i;

            futures.add(executor.submit(new Callable<AtomicBoolean>() {
                @Override
                public AtomicBoolean call() throws Exception {
                    barrier.await();

                    final AtomicBoolean moveStatus = new AtomicBoolean(false);
                    CountDownLatch countDownLatch = new CountDownLatch(1);
                    final PositionImpl resetPosition = new PositionImpl(lastPosition.getLedgerId(),
                            lastPosition.getEntryId() - (5 * idx));

                    cursor.asyncResetCursor(resetPosition, false, new AsyncCallbacks.ResetCursorCallback() {
                        @Override
                        public void resetComplete(Object ctx) {
                            moveStatus.set(true);
                            PositionImpl pos = (PositionImpl) ctx;
                            log.info("move to [{}] completed for consumer [{}]", pos.toString(), idx);
                            countDownLatch.countDown();
                        }

                        @Override
                        public void resetFailed(ManagedLedgerException exception, Object ctx) {
                            moveStatus.set(false);
                            PositionImpl pos = (PositionImpl) ctx;
                            log.warn("move to [{}] failed for consumer [{}]", pos.toString(), idx);
                            countDownLatch.countDown();
                        }
                    });
                    countDownLatch.await();
                    assertEquals(resetPosition, cursor.getReadPosition());
                    cursor.close();

                    return moveStatus;
                }
            }));
        }

        barrier.await();

        for (Future<AtomicBoolean> f : futures) {
            assertTrue(f.get().get());
        }
        ledger.close();
    }

    @Test(timeOut = 20000)
    void testLastActiveAfterResetCursor() throws Exception {
        ManagedLedger ledger = factory.open("test_cursor_ledger");
        ManagedCursor cursor = ledger.openCursor("tla");

        PositionImpl lastPosition = null;
        for (int i = 0; i < 3; i++) {
            lastPosition = (PositionImpl) ledger.addEntry("dummy-entry".getBytes(Encoding));
        }

        final AtomicBoolean moveStatus = new AtomicBoolean(false);
        CountDownLatch countDownLatch = new CountDownLatch(1);

        long lastActive = cursor.getLastActive();

        cursor.asyncResetCursor(lastPosition, false, new AsyncCallbacks.ResetCursorCallback() {
            @Override
            public void resetComplete(Object ctx) {
                moveStatus.set(true);
                countDownLatch.countDown();
            }

            @Override
            public void resetFailed(ManagedLedgerException exception, Object ctx) {
                moveStatus.set(false);
                countDownLatch.countDown();
            }
        });

        countDownLatch.await();
        assertTrue(moveStatus.get());

        assertNotNull(lastPosition);
        assertEquals(lastPosition, cursor.getReadPosition());

        assertNotEquals(lastActive, cursor.getLastActive());

        cursor.close();
        ledger.close();
    }

    @Test(timeOut = 20000)
    void seekPosition() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(10));
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));

        cursor.seek(new PositionImpl(lastPosition.getLedgerId(), lastPosition.getEntryId() - 1));
    }

    @Test(timeOut = 20000)
    void seekPosition2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        PositionImpl seekPosition = (PositionImpl) ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        cursor.seek(new PositionImpl(seekPosition.getLedgerId(), seekPosition.getEntryId()));
    }

    @Test(timeOut = 20000)
    void seekPosition3() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl seekPosition = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        Position entry5 = ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        Position entry6 = ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        cursor.seek(new PositionImpl(seekPosition.getLedgerId(), seekPosition.getEntryId()));

        assertEquals(cursor.getReadPosition(), seekPosition);
        List<Entry> entries = cursor.readEntries(1);
        assertEquals(entries.size(), 1);
        assertEquals(new String(entries.get(0).getData(), Encoding), "dummy-entry-4");
        entries.forEach(e -> e.release());

        cursor.seek(entry5.getNext());
        assertEquals(cursor.getReadPosition(), entry6);
        entries = cursor.readEntries(1);
        assertEquals(entries.size(), 1);
        assertEquals(new String(entries.get(0).getData(), Encoding), "dummy-entry-6");
        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void seekPosition4() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("c1");
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        cursor.markDelete(p1);
        assertEquals(cursor.getMarkDeletedPosition(), p1);
        assertEquals(cursor.getReadPosition(), p2);

        List<Entry> entries = cursor.readEntries(2);
        entries.forEach(e -> e.release());

        cursor.seek(p2);
        assertEquals(cursor.getMarkDeletedPosition(), p1);
        assertEquals(cursor.getReadPosition(), p2);
    }

    @Test(timeOut = 20000)
    void rewind() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
        ManagedCursor c1 = ledger.openCursor("c1");
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        Position p3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        Position p4 = ledger.addEntry("dummy-entry-4".getBytes(Encoding));

        log.debug("p1: {}", p1);
        log.debug("p2: {}", p2);
        log.debug("p3: {}", p3);
        log.debug("p4: {}", p4);

        assertEquals(c1.getNumberOfEntries(), 4);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 4);
        c1.markDelete(p1);
        assertEquals(c1.getNumberOfEntries(), 3);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        List<Entry> entries = c1.readEntries(10);
        assertEquals(entries.size(), 3);
        entries.forEach(e -> e.release());

        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        c1.rewind();
        assertEquals(c1.getNumberOfEntries(), 3);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        c1.markDelete(p2);
        assertEquals(c1.getNumberOfEntries(), 2);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);

        entries = c1.readEntries(10);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());

        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);
        c1.rewind();
        assertEquals(c1.getNumberOfEntries(), 2);
        c1.markDelete(p4);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
        c1.rewind();
        assertEquals(c1.getNumberOfEntries(), 0);
        ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        assertEquals(c1.getNumberOfEntries(), 1);
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));
        assertEquals(c1.getNumberOfEntries(), 2);
    }

    @Test(timeOut = 20000)
    void markDeleteSkippingMessage() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(10));
        ManagedCursor cursor = ledger.openCursor("c1");
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        PositionImpl p4 = (PositionImpl) ledger.addEntry("dummy-entry-4".getBytes(Encoding));

        assertEquals(cursor.getNumberOfEntries(), 4);

        cursor.markDelete(p1);
        assertTrue(cursor.hasMoreEntries());
        assertEquals(cursor.getNumberOfEntries(), 3);

        assertEquals(cursor.getReadPosition(), p2);

        List<Entry> entries = cursor.readEntries(1);
        assertEquals(entries.size(), 1);
        assertEquals(new String(entries.get(0).getData(), Encoding), "dummy-entry-2");
        entries.forEach(e -> e.release());

        cursor.markDelete(p4);
        assertFalse(cursor.hasMoreEntries());
        assertEquals(cursor.getNumberOfEntries(), 0);

        assertEquals(cursor.getReadPosition(), new PositionImpl(p4.getLedgerId(), p4.getEntryId() + 1));
    }

    @Test(timeOut = 20000)
    void removingCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));
        ManagedCursor cursor = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        assertEquals(cursor.getNumberOfEntries(), 6);
        assertEquals(ledger.getNumberOfEntries(), 6);
        ledger.deleteCursor("c1");

        // Verify that it's a new empty cursor
        cursor = ledger.openCursor("c1");
        assertEquals(cursor.getNumberOfEntries(), 0);
        ledger.addEntry("dummy-entry-7".getBytes(Encoding));

        // Verify that GC trimming kicks in
        Awaitility.await().until(() -> ledger.getNumberOfEntries() <= 2);
    }

    @Test(timeOut = 20000)
    void cursorPersistence() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor c1 = ledger.openCursor("c1");
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(3);
        Position p1 = entries.get(2).getPosition();
        c1.markDelete(p1);
        entries.forEach(e -> e.release());

        entries = c1.readEntries(4);
        Position p2 = entries.get(2).getPosition();
        c2.markDelete(p2);
        entries.forEach(e -> e.release());

        // Reopen

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger");
        c1 = ledger.openCursor("c1");
        c2 = ledger.openCursor("c2");

        assertEquals(c1.getMarkDeletedPosition(), p1);
        assertEquals(c2.getMarkDeletedPosition(), p2);
    }

    @Test(timeOut = 20000)
    void cursorPersistence2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger",
                new ManagedLedgerConfig().setMetadataMaxEntriesPerLedger(1));
        ManagedCursor c1 = ledger.openCursor("c1");
        ManagedCursor c2 = ledger.openCursor("c2");
        ManagedCursor c3 = ledger.openCursor("c3");
        Position p0 = c3.getMarkDeletedPosition();
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c4 = ledger.openCursor("c4");
        Position p2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        Position p3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        Position p4 = ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        Position p5 = ledger.addEntry("dummy-entry-5".getBytes(Encoding));
        ledger.addEntry("dummy-entry-6".getBytes(Encoding));

        c1.markDelete(p1);
        c1.markDelete(p2);
        c1.markDelete(p3);
        c1.markDelete(p4);
        c1.markDelete(p5);

        c2.markDelete(p1);

        // Reopen

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger");
        c1 = ledger.openCursor("c1");
        c2 = ledger.openCursor("c2");
        c3 = ledger.openCursor("c3");
        c4 = ledger.openCursor("c4");

        assertEquals(c1.getMarkDeletedPosition(), p5);
        assertEquals(c2.getMarkDeletedPosition(), p1);
        assertEquals(c3.getMarkDeletedPosition(), p0);
        assertEquals(c4.getMarkDeletedPosition(), p1);
    }

    @Test
    public void asyncMarkDeleteBlocking() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setMaxEntriesPerLedger(10);
        config.setMetadataMaxEntriesPerLedger(5);
        ManagedLedger ledger = factory.open("my_test_ledger", config);
        final ManagedCursor c1 = ledger.openCursor("c1");
        final AtomicReference<Position> lastPosition = new AtomicReference<Position>();

        final int N = 100;
        final CountDownLatch latch = new CountDownLatch(N);
        for (int i = 0; i < N; i++) {
            ledger.asyncAddEntry("entry".getBytes(Encoding), new AddEntryCallback() {
                @Override
                public void addFailed(ManagedLedgerException exception, Object ctx) {
                }

                @Override
                public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                    lastPosition.set(position);
                    c1.asyncMarkDelete(position, new MarkDeleteCallback() {
                        @Override
                        public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                        }

                        @Override
                        public void markDeleteComplete(Object ctx) {
                            latch.countDown();
                        }
                    }, null);
                }
            }, null);
        }

        latch.await();

        assertEquals(c1.getNumberOfEntries(), 0);

        // Reopen
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger");
        ManagedCursor c2 = ledger.openCursor("c1");

        assertEquals(c2.getMarkDeletedPosition(), lastPosition.get());
    }

    @Test(timeOut = 20000)
    void cursorPersistenceAsyncMarkDeleteSameThread() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger",
                new ManagedLedgerConfig().setMetadataMaxEntriesPerLedger(5));
        final ManagedCursor c1 = ledger.openCursor("c1");

        final int N = 100;
        List<Position> positions = Lists.newArrayList();
        for (int i = 0; i < N; i++) {
            Position p = ledger.addEntry("dummy-entry".getBytes(Encoding));
            positions.add(p);
        }

        Position lastPosition = positions.get(N - 1);

        final CountDownLatch latch = new CountDownLatch(N);
        for (final Position p : positions) {
            c1.asyncMarkDelete(p, new MarkDeleteCallback() {
                @Override
                public void markDeleteComplete(Object ctx) {
                    latch.countDown();
                }

                @Override
                public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                    log.error("Failed to markdelete", exception);
                    latch.countDown();
                }
            }, null);
        }

        latch.await();

        // Reopen
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger");
        ManagedCursor c2 = ledger.openCursor("c1");

        assertEquals(c2.getMarkDeletedPosition(), lastPosition);
    }

    @Test(timeOut = 20000)
    void unorderedMarkDelete() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        final ManagedCursor c1 = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("entry-2".getBytes(Encoding));

        c1.markDelete(p2);
        try {
            c1.markDelete(p1);
            fail("Should have thrown exception");
        } catch (ManagedLedgerException e) {
            // ok
        }

        assertEquals(c1.getMarkDeletedPosition(), p2);
    }

    @Test(timeOut = 20000)
    void unorderedAsyncMarkDelete() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        final ManagedCursor c1 = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("entry-2".getBytes(Encoding));

        final CountDownLatch latch = new CountDownLatch(2);
        c1.asyncMarkDelete(p2, new MarkDeleteCallback() {
            @Override
            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                fail();
            }

            @Override
            public void markDeleteComplete(Object ctx) {
                latch.countDown();
            }
        }, null);

        c1.asyncMarkDelete(p1, new MarkDeleteCallback() {
            @Override
            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                latch.countDown();
            }

            @Override
            public void markDeleteComplete(Object ctx) {
                fail();
            }
        }, null);

        latch.await();

        assertEquals(c1.getMarkDeletedPosition(), p2);
    }

    @Test(timeOut = 20000)
    void deleteCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor c1 = ledger.openCursor("c1");

        ledger.addEntry("entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("entry-2".getBytes(Encoding));

        assertEquals(c1.getNumberOfEntries(), 2);

        // Remove and recreate the same cursor
        ledger.deleteCursor("c1");

        try {
            c1.readEntries(10);
            fail("must fail, the cursor should be closed");
        } catch (ManagedLedgerException e) {
            // ok
        }

        try {
            c1.markDelete(p2);
            fail("must fail, the cursor should be closed");
        } catch (ManagedLedgerException e) {
            // ok
        }

        c1 = ledger.openCursor("c1");
        assertEquals(c1.getNumberOfEntries(), 0);

        c1.close();
        try {
            c1.readEntries(10);
            fail("must fail, the cursor should be closed");
        } catch (ManagedLedgerException e) {
            // ok
        }

        c1.close();
    }

    @Test(timeOut = 20000)
    void errorCreatingCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        bkc.failAfter(1, BKException.Code.NotEnoughBookiesException);
        metadataStore.failConditional(new MetadataStoreException("error"), (op, path) ->
                path.equals("/managed-ledgers/my_test_ledger/c1")
                        && op == FaultInjectionMetadataStore.OperationType.PUT
        );

        try {
            ledger.openCursor("c1");
            fail("should have failed");
        } catch (ManagedLedgerException e) {
            // ok
        }
    }

    @Test
    void failDuringRecoveryWithEmptyLedger() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursor cursor = ledger.openCursor("cursor");

        ledger.addEntry("entry-1".getBytes());
        Position p2 = ledger.addEntry("entry-2".getBytes());
        Position p3 = ledger.addEntry("entry-3".getBytes());

        cursor.markDelete(p2);
        // Do graceful close so snapshot is forced
        ledger.close();

        // Re-open
        ledger = factory.open("my_test_ledger");
        cursor = ledger.openCursor("cursor");
        cursor.markDelete(p3);

        // Force-reopen so the recovery will be forced to read from ledger
        bkc.returnEmptyLedgerAfter(1);
        ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig();

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, conf);
        ledger = factory2.open("my_test_ledger");
        cursor = ledger.openCursor("cursor");

        // Cursor was rolled back to p2 because of the ledger recovery failure
        assertEquals(cursor.getMarkDeletedPosition(), p2);
    }

    @Test(timeOut = 20000)
    void errorRecoveringCursor() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        Position p1 = ledger.addEntry("entry".getBytes());
        ledger.addEntry("entry".getBytes());
        ManagedCursor c1 = ledger.openCursor("c1");
        Position p3 = ledger.addEntry("entry".getBytes());

        assertEquals(c1.getReadPosition(), p3);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);

        bkc.failAfter(3, BKException.Code.LedgerRecoveryException);

        ledger = factory2.open("my_test_ledger");
        c1 = ledger.openCursor("c1");

        // Verify the ManagedCursor was rewind back to the snapshotted position
        assertEquals(c1.getReadPosition(), p3);
    }

    @Test(timeOut = 20000)
    void errorRecoveringCursor2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ledger.openCursor("c1");

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);

        bkc.failAfter(4, BKException.Code.MetadataVersionException);

        try {
            ledger = factory2.open("my_test_ledger");
            fail("should have failed");
        } catch (ManagedLedgerException e) {
            // ok
        }
    }

    @Test(timeOut = 20000)
    void errorRecoveringCursor3() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        Position p1 = ledger.addEntry("entry".getBytes());
        ledger.addEntry("entry".getBytes());
        ManagedCursor c1 = ledger.openCursor("c1");
        Position p3 = ledger.addEntry("entry".getBytes());

        assertEquals(c1.getReadPosition(), p3);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);

        bkc.failAfter(4, BKException.Code.ReadException);

        ledger = factory2.open("my_test_ledger");
        c1 = ledger.openCursor("c1");

        // Verify the ManagedCursor was rewind back to the snapshotted position
        assertEquals(c1.getReadPosition(), p3);
    }

    @Test(timeOut = 20000)
    void testSingleDelete() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(3));
        ManagedCursor cursor = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry1".getBytes());
        Position p2 = ledger.addEntry("entry2".getBytes());
        Position p3 = ledger.addEntry("entry3".getBytes());
        Position p4 = ledger.addEntry("entry4".getBytes());
        Position p5 = ledger.addEntry("entry5".getBytes());
        Position p6 = ledger.addEntry("entry6".getBytes());

        Position p0 = cursor.getMarkDeletedPosition();

        cursor.delete(p4);
        assertEquals(cursor.getMarkDeletedPosition(), p0);

        cursor.delete(p1);
        assertEquals(cursor.getMarkDeletedPosition(), p1);

        cursor.delete(p3);

        // Delete will silently succeed
        cursor.delete(p3);
        assertEquals(cursor.getMarkDeletedPosition(), p1);

        cursor.delete(p2);
        assertEquals(cursor.getMarkDeletedPosition(), p4);

        cursor.delete(p5);
        assertEquals(cursor.getMarkDeletedPosition(), p5);

        cursor.close();
        try {
            cursor.delete(p6);
        } catch (ManagedLedgerException e) {
            // Ok
        }
    }

    @Test(timeOut = 20000)
    void testFilteringReadEntries() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(3));
        ManagedCursor cursor = ledger.openCursor("c1");

        /* Position p1 = */ledger.addEntry("entry1".getBytes());
        /* Position p2 = */ledger.addEntry("entry2".getBytes());
        /* Position p3 = */ledger.addEntry("entry3".getBytes());
        /* Position p4 = */ledger.addEntry("entry4".getBytes());
        Position p5 = ledger.addEntry("entry5".getBytes());
        /* Position p6 = */ledger.addEntry("entry6".getBytes());

        assertEquals(cursor.getNumberOfEntries(), 6);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 6);

        List<Entry> entries = cursor.readEntries(3);
        assertEquals(entries.size(), 3);
        entries.forEach(e -> e.release());

        assertEquals(cursor.getNumberOfEntries(), 3);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 6);

        log.info("Deleting {}", p5);
        cursor.delete(p5);

        assertEquals(cursor.getNumberOfEntries(), 2);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 5);

        entries = cursor.readEntries(3);
        assertEquals(entries.size(), 2);
        entries.forEach(e -> e.release());
        assertEquals(cursor.getNumberOfEntries(), 0);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 5);
    }

    @Test(timeOut = 20000)
    void testReadingAllFilteredEntries() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(3));
        ledger.openCursor("c1");
        ManagedCursor c2 = ledger.openCursor("c2");

        ledger.addEntry("entry1".getBytes());
        Position p2 = ledger.addEntry("entry2".getBytes());
        Position p3 = ledger.addEntry("entry3".getBytes());
        Position p4 = ledger.addEntry("entry4".getBytes());
        Position p5 = ledger.addEntry("entry5".getBytes());

        c2.readEntries(1).get(0).release();
        c2.delete(p2);
        c2.delete(p3);

        List<Entry> entries = c2.readEntries(2);
        assertEquals(entries.size(), 2);
        assertEquals(entries.get(0).getPosition(), p4);
        assertEquals(entries.get(1).getPosition(), p5);
        entries.forEach(e -> e.release());
    }

    @Test(timeOut = 20000)
    void testCountingWithDeletedEntries() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
        ManagedCursor cursor = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry1".getBytes());
        /* Position p2 = */ledger.addEntry("entry2".getBytes());
        /* Position p3 = */ledger.addEntry("entry3".getBytes());
        /* Position p4 = */ledger.addEntry("entry4".getBytes());
        Position p5 = ledger.addEntry("entry5".getBytes());
        Position p6 = ledger.addEntry("entry6".getBytes());
        Position p7 = ledger.addEntry("entry7".getBytes());
        Position p8 = ledger.addEntry("entry8".getBytes());

        assertEquals(cursor.getNumberOfEntries(), 8);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 8);

        cursor.delete(p8);
        assertEquals(cursor.getNumberOfEntries(), 7);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 7);

        cursor.delete(p1);
        assertEquals(cursor.getNumberOfEntries(), 6);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 6);

        cursor.delete(p7);
        cursor.delete(p6);
        cursor.delete(p5);
        assertEquals(cursor.getNumberOfEntries(), 3);
        assertEquals(cursor.getNumberOfEntriesInBacklog(false), 3);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testMarkDeleteTwice(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
                .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(2));
        ManagedCursor cursor = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry1".getBytes());
        cursor.markDelete(p1);
        cursor.markDelete(p1);

        assertEquals(cursor.getMarkDeletedPosition(), p1);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testSkipEntries(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
                .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(2));
        Position pos;

        ManagedCursor c1 = ledger.openCursor("c1");

        // test skip on empty ledger
        pos = c1.getReadPosition();
        c1.skipEntries(1, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getReadPosition(), pos);

        pos = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        pos = ledger.addEntry("dummy-entry-2".getBytes(Encoding));

        // skip entries in same ledger
        c1.skipEntries(1, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getNumberOfEntries(), 1);

        // skip entries until end of ledger
        c1.skipEntries(1, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getReadPosition(), pos.getNext());
        assertEquals(c1.getMarkDeletedPosition(), pos);

        // skip entries across ledgers
        for (int i = 0; i < 6; i++) {
            pos = ledger.addEntry("dummy-entry".getBytes(Encoding));
        }

        c1.skipEntries(5, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getNumberOfEntries(), 1);

        // skip more than the current set of entries
        c1.skipEntries(10, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertFalse(c1.hasMoreEntries());
        assertEquals(c1.getReadPosition(), pos.getNext());
        assertEquals(c1.getMarkDeletedPosition(), pos);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testSkipEntriesWithIndividualDeletedMessages(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testSkipEntriesWithIndividualDeletedMessages", new ManagedLedgerConfig()
                .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(5));
        ManagedCursor c1 = ledger.openCursor("c1");

        Position pos1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        Position pos2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        Position pos3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        Position pos4 = ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        Position pos5 = ledger.addEntry("dummy-entry-5".getBytes(Encoding));

        // delete individual messages
        c1.delete(pos2);
        c1.delete(pos4);

        c1.skipEntries(3, IndividualDeletedEntries.Exclude);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getReadPosition(), pos5.getNext());
        assertEquals(c1.getMarkDeletedPosition(), pos5);

        pos1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        pos2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        pos3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));
        pos4 = ledger.addEntry("dummy-entry-4".getBytes(Encoding));
        pos5 = ledger.addEntry("dummy-entry-5".getBytes(Encoding));

        c1.delete(pos2);
        c1.delete(pos4);

        c1.skipEntries(4, IndividualDeletedEntries.Include);
        assertEquals(c1.getNumberOfEntries(), 1);
        assertEquals(c1.getReadPosition(), pos5);
        assertEquals(c1.getMarkDeletedPosition(), pos4);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testClearBacklog(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
                .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ManagedCursor c2 = ledger.openCursor("c2");
        ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        ManagedCursor c3 = ledger.openCursor("c3");
        ledger.addEntry("dummy-entry-3".getBytes(Encoding));

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        assertEquals(c1.getNumberOfEntries(), 3);
        assertTrue(c1.hasMoreEntries());

        c1.clearBacklog();
        c3.clearBacklog();

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertFalse(c1.hasMoreEntries());

        assertEquals(c2.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c2.getNumberOfEntries(), 2);
        assertTrue(c2.hasMoreEntries());

        assertEquals(c3.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c3.getNumberOfEntries(), 0);
        assertFalse(c3.hasMoreEntries());

        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        c1 = ledger.openCursor("c1");
        c2 = ledger.openCursor("c2");
        c3 = ledger.openCursor("c3");

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertFalse(c1.hasMoreEntries());

        assertEquals(c2.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c2.getNumberOfEntries(), 2);
        assertTrue(c2.hasMoreEntries());

        assertEquals(c3.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c3.getNumberOfEntries(), 0);
        assertFalse(c3.hasMoreEntries());
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testRateLimitMarkDelete(boolean useOpenRangeSet) throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setThrottleMarkDelete(1).setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet); // Throttle to 1/s
        ManagedLedger ledger = factory.open("my_test_ledger", config);

        ManagedCursor c1 = ledger.openCursor("c1");
        Position p1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("dummy-entry-2".getBytes(Encoding));
        Position p3 = ledger.addEntry("dummy-entry-3".getBytes(Encoding));

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        c1.markDelete(p1);
        c1.markDelete(p2);
        c1.markDelete(p3);

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);

        // Re-open to recover from storage
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig());

        c1 = ledger.openCursor("c1");

        // Only the 1st mark-delete was persisted
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void deleteSingleMessageTwice(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        ManagedCursor c1 = ledger.openCursor("c1");

        Position p1 = ledger.addEntry("entry-1".getBytes(Encoding));
        Position p2 = ledger.addEntry("entry-2".getBytes(Encoding));
        Position p3 = ledger.addEntry("entry-3".getBytes(Encoding));
        Position p4 = ledger.addEntry("entry-4".getBytes(Encoding));

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 4);
        assertEquals(c1.getNumberOfEntries(), 4);

        c1.delete(p1);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        assertEquals(c1.getNumberOfEntries(), 3);
        assertEquals(c1.getMarkDeletedPosition(), p1);
        assertEquals(c1.getReadPosition(), p2);

        // Should have not effect since p1 is already deleted
        c1.delete(p1);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 3);
        assertEquals(c1.getNumberOfEntries(), 3);
        assertEquals(c1.getMarkDeletedPosition(), p1);
        assertEquals(c1.getReadPosition(), p2);

        c1.delete(p2);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c1.getNumberOfEntries(), 2);
        assertEquals(c1.getMarkDeletedPosition(), p2);
        assertEquals(c1.getReadPosition(), p3);

        // Should have not effect since p2 is already deleted
        c1.delete(p2);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 2);
        assertEquals(c1.getNumberOfEntries(), 2);
        assertEquals(c1.getMarkDeletedPosition(), p2);
        assertEquals(c1.getReadPosition(), p3);

        c1.delete(p3);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 1);
        assertEquals(c1.getNumberOfEntries(), 1);
        assertEquals(c1.getMarkDeletedPosition(), p3);
        assertEquals(c1.getReadPosition(), p4);

        // Should have not effect since p3 is already deleted
        c1.delete(p3);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 1);
        assertEquals(c1.getNumberOfEntries(), 1);
        assertEquals(c1.getMarkDeletedPosition(), p3);
        assertEquals(c1.getReadPosition(), p4);

        c1.delete(p4);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getMarkDeletedPosition(), p4);
        assertEquals(c1.getReadPosition(), p4.getNext());

        // Should have not effect since p4 is already deleted
        c1.delete(p4);
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 0);
        assertEquals(c1.getNumberOfEntries(), 0);
        assertEquals(c1.getMarkDeletedPosition(), p4);
        assertEquals(c1.getReadPosition(), p4.getNext());
    }

    @Test(timeOut = 10000, dataProvider = "useOpenRangeSet")
    void testReadEntriesOrWait(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        final int Consumers = 10;
        final CountDownLatch counter = new CountDownLatch(Consumers);

        for (int i = 0; i < Consumers; i++) {
            ManagedCursor c = ledger.openCursor("c" + i);

            c.asyncReadEntriesOrWait(1, new ReadEntriesCallback() {
                @Override
                public void readEntriesComplete(List<Entry> entries, Object ctx) {
                    assertEquals(entries.size(), 1);
                    entries.forEach(e -> e.release());
                    counter.countDown();
                }

                @Override
                public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                    log.error("Error reading", exception);
                }
            }, null, PositionImpl.latest);
        }

        ledger.addEntry("test".getBytes());
        counter.await();
    }

    @Test(timeOut = 20000)
    void testReadEntriesOrWaitBlocking() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        final int Messages = 100;
        final int Consumers = 10;

        List<Future<Void>> futures = Lists.newArrayList();
        @Cleanup("shutdownNow")
        ExecutorService executor = Executors.newCachedThreadPool();
        final CyclicBarrier barrier = new CyclicBarrier(Consumers + 1);

        for (int i = 0; i < Consumers; i++) {
            final ManagedCursor cursor = ledger.openCursor("c" + i);

            futures.add(executor.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    barrier.await();

                    int toRead = Messages;
                    while (toRead > 0) {
                        List<Entry> entries = cursor.readEntriesOrWait(10);
                        assertTrue(entries.size() <= 10);
                        toRead -= entries.size();
                        entries.forEach(e -> e.release());
                    }

                    return null;
                }
            }));
        }

        barrier.await();
        for (int i = 0; i < Messages; i++) {
            ledger.addEntry("test".getBytes());
        }

        for (Future<Void> f : futures) {
            f.get();
        }
    }

    @Test(timeOut = 20000)
    void testFindNewestMatching() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertNull(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))));
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingOdd1() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        Position p1 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p1);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingOdd2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        Position p2 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p2);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingOdd3() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position p3 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p3);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingOdd4() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position p4 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p4);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingOdd5() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position p5 = ledger.addEntry("expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p5);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEven1() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        Position p1 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p1);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEven2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        Position p2 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p2);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEven3() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position p3 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p3);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEven4() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");

        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position p4 = ledger.addEntry("expired".getBytes(Encoding));

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p4);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase1() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        assertNull(c1.findNewestMatching(
            entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))));
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase2() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        Position p1 = ledger.addEntry("expired".getBytes(Encoding));
        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p1);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase3() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        Position p1 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p1);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase4() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        Position p1 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p1);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase5() throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase5");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        Position p2 = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                p2);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testFindNewestMatchingEdgeCase6(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase6", new ManagedLedgerConfig()
                .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(3));

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position newPosition = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));
        List<Entry> entries = c1.readEntries(3);
        c1.markDelete(entries.get(2).getPosition());
        entries.forEach(e -> e.release());
        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                newPosition);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testFindNewestMatchingEdgeCase7(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase7",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position lastPosition = ledger.addEntry("expired".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.markDelete(entries.get(0).getPosition());
        c1.delete(entries.get(2).getPosition());
        entries.forEach(e -> e.release());

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                lastPosition);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase8() throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase8");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position lastPosition = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(2).getPosition());
        entries.forEach(e -> e.release());

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                lastPosition);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase9() throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase9");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position lastPosition = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(5);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(3).getPosition());
        entries.forEach(e -> e.release());

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                lastPosition);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingEdgeCase10() throws Exception {
        ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase10");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("expired".getBytes(Encoding));
        Position lastPosition = ledger.addEntry("expired".getBytes(Encoding));
        ledger.addEntry("not-expired".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(7);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(3).getPosition());
        c1.delete(entries.get(6).getPosition());
        entries.forEach(e -> e.release());

        assertEquals(
                c1.findNewestMatching(entry -> Arrays.equals(entry.getDataAndRelease(), "expired".getBytes(Encoding))),
                lastPosition);
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testIndividuallyDeletedMessages(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("entry-0".getBytes(Encoding));
        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));
        ledger.addEntry("entry-3".getBytes(Encoding));
        ledger.addEntry("entry-4".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(2).getPosition());
        c1.markDelete(entries.get(3).getPosition());
        entries.forEach(e -> e.release());

        assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
    }

    @Test(timeOut = 20000)
    void testIndividuallyDeletedMessages1() throws Exception {
        ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages1");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("entry-0".getBytes(Encoding));
        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));
        ledger.addEntry("entry-3".getBytes(Encoding));
        ledger.addEntry("entry-4".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.delete(entries.get(1).getPosition());
        c1.markDelete(entries.get(3).getPosition());
        entries.forEach(e -> e.release());

        assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testIndividuallyDeletedMessages2(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages2",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("entry-0".getBytes(Encoding));
        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));
        ledger.addEntry("entry-3".getBytes(Encoding));
        ledger.addEntry("entry-4".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(2).getPosition());
        c1.delete(entries.get(0).getPosition());
        entries.forEach(e -> e.release());

        assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
    }

    @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
    void testIndividuallyDeletedMessages3(boolean useOpenRangeSet) throws Exception {
        ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages3",
                new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("entry-0".getBytes(Encoding));
        ledger.addEntry("entry-1".getBytes(Encoding));
        ledger.addEntry("entry-2".getBytes(Encoding));
        ledger.addEntry("entry-3".getBytes(Encoding));
        ledger.addEntry("entry-4".getBytes(Encoding));

        List<Entry> entries = c1.readEntries(4);
        c1.delete(entries.get(1).getPosition());
        c1.delete(entries.get(2).getPosition());
        c1.markDelete(entries.get(0).getPosition());
        entries.forEach(e -> e.release());

        assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingAfterLedgerRollover() throws Exception {
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("my_test_ledger");
        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        ledger.addEntry("first-expired".getBytes(Encoding));
        ledger.addEntry("second".getBytes(Encoding));
        ledger.addEntry("third".getBytes(Encoding));
        ledger.addEntry("fourth".getBytes(Encoding));
        Position last = ledger.addEntry("last-expired".getBytes(Encoding));

        // roll a new ledger
        int numLedgersBefore = ledger.getLedgersInfo().size();
        ledger.getConfig().setMaxEntriesPerLedger(1);
        Field stateUpdater = ManagedLedgerImpl.class.getDeclaredField("state");
        stateUpdater.setAccessible(true);
        stateUpdater.set(ledger, ManagedLedgerImpl.State.LedgerOpened);
        ledger.rollCurrentLedgerIfFull();
        Awaitility.await().atMost(20, TimeUnit.SECONDS)
                .until(() -> ledger.getLedgersInfo().size() > numLedgersBefore);

        // the algorithm looks for "expired" messages
        // starting from the first, then it moves to the last message
        // if the condition evaluates to true on the last message
        // then we are done
        // there was a bug (https://github.com/apache/pulsar/issues/9082)
        // in which if the last message was in a different ledger
        // the jump from the first message to the last message went
        // to an invalid position and so the search stopped at the first message

        // we want to assert here that the algorithm returns the position of the
        // last message
        assertEquals(last,
                c1.findNewestMatching(entry -> {
                    byte[] data = entry.getDataAndRelease();
                    return Arrays.equals(data, "first-expired".getBytes(Encoding))
                            || Arrays.equals(data, "last-expired".getBytes(Encoding));
                }));

    }

    public static byte[] getEntryPublishTime(String msg) throws Exception {
        return Long.toString(System.currentTimeMillis()).getBytes();
    }

    public Position findPositionFromAllEntries(ManagedCursor c1, final long timestamp) throws Exception {
        final CountDownLatch counter = new CountDownLatch(1);
        class Result {
            ManagedLedgerException exception = null;
            Position position = null;
        }

        final Result result = new Result();
        AsyncCallbacks.FindEntryCallback findEntryCallback = new AsyncCallbacks.FindEntryCallback() {
            @Override
            public void findEntryComplete(Position position, Object ctx) {
                result.position = position;
                counter.countDown();
            }

            @Override
            public void findEntryFailed(ManagedLedgerException exception, Optional<Position> failedReadPosition,
                    Object ctx) {
                result.exception = exception;
                counter.countDown();
            }
        };

        c1.asyncFindNewestMatching(ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries, entry -> {

            try {
                long publishTime = Long.parseLong(new String(entry.getData()));
                return publishTime <= timestamp;
            } catch (Exception e) {
                log.error("Error de-serializing message for message position find", e);
            } finally {
                entry.release();
            }
            return false;
        }, findEntryCallback, ManagedCursorImpl.FindPositionConstraint.SearchAllAvailableEntries);
        counter.await();
        if (result.exception != null) {
            throw result.exception;
        }
        return result.position;
    }

    void internalTestFindNewestMatchingAllEntries(final String name, final int entriesPerLedger,
            final int expectedEntryId) throws Exception {
        final String ledgerAndCursorName = name;
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setRetentionSizeInMB(10);
        config.setMaxEntriesPerLedger(entriesPerLedger);
        config.setRetentionTime(1, TimeUnit.HOURS);
        ManagedLedger ledger = factory.open(ledgerAndCursorName, config);
        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);

        ledger.addEntry(getEntryPublishTime("retained1"));
        // space apart message publish times
        Thread.sleep(100);
        ledger.addEntry(getEntryPublishTime("retained2"));
        Thread.sleep(100);
        ledger.addEntry(getEntryPublishTime("retained3"));
        Thread.sleep(100);
        Position newPosition = ledger.addEntry(getEntryPublishTime("expectedresetposition"));
        long timestamp = System.currentTimeMillis();
        long ledgerId = ((PositionImpl) newPosition).getLedgerId();
        Thread.sleep(2);

        ledger.addEntry(getEntryPublishTime("not-read"));
        List<Entry> entries = c1.readEntries(3);
        c1.markDelete(entries.get(2).getPosition());
        c1.close();
        ledger.close();
        entries.forEach(e -> e.release());
        // give timed ledger trimming a chance to run
        Thread.sleep(100);

        ledger = factory.open(ledgerAndCursorName, config);
        c1 = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);

        PositionImpl found = (PositionImpl) findPositionFromAllEntries(c1, timestamp);
        assertEquals(found.getLedgerId(), ledgerId);
        assertEquals(found.getEntryId(), expectedEntryId);

        found = (PositionImpl) findPositionFromAllEntries(c1, 0);
        assertNull(found);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingAllEntries() throws Exception {
        final String ledgerAndCursorName = "testFindNewestMatchingAllEntries";
        // condition below assumes entries per ledger is 2
        // needs to be changed if entries per ledger is changed
        int expectedEntryId = 1;
        int entriesPerLedger = 2;
        internalTestFindNewestMatchingAllEntries(ledgerAndCursorName, entriesPerLedger, expectedEntryId);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingAllEntries2() throws Exception {
        final String ledgerAndCursorName = "testFindNewestMatchingAllEntries2";
        // condition below assumes entries per ledger is 1
        // needs to be changed if entries per ledger is changed
        int expectedEntryId = 0;
        int entriesPerLedger = 1;
        internalTestFindNewestMatchingAllEntries(ledgerAndCursorName, entriesPerLedger, expectedEntryId);
    }

    @Test(timeOut = 20000)
    void testFindNewestMatchingAllEntriesSingleLedger() throws Exception {
        final String ledgerAndCursorName = "testFindNewestMatchingAllEntriesSingleLedger";
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        // needs to be changed if entries per ledger is changed
        int expectedEntryId = 3;
        int entriesPerLedger = config.getMaxEntriesPerLedger();
        internalTestFindNewestMatchingAllEntries(ledgerAndCursorName, entriesPerLedger, expectedEntryId);
    }

    @Test(timeOut = 20000)
    void testReplayEntries() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        PositionImpl p1 = (PositionImpl) ledger.addEntry("entry1".getBytes(Encoding));
        PositionImpl p2 = (PositionImpl) ledger.addEntry("entry2".getBytes(Encoding));
        PositionImpl p3 = (PositionImpl) ledger.addEntry("entry3".getBytes(Encoding));
        ledger.addEntry("entry4".getBytes(Encoding));

        // 1. Replay empty position set should return empty entry set
        Set<PositionImpl> positions = Sets.newHashSet();
        assertTrue(c1.replayEntries(positions).isEmpty());

        positions.add(p1);
        positions.add(p3);

        // 2. entries 1 and 3 should be returned, but they can be in any order
        List<Entry> entries = c1.replayEntries(positions);
        assertEquals(entries.size(), 2);
        assertTrue((Arrays.equals(entries.get(0).getData(), "entry1".getBytes(Encoding))
                && Arrays.equals(entries.get(1).getData(), "entry3".getBytes(Encoding)))
                || (Arrays.equals(entries.get(0).getData(), "entry3".getBytes(Encoding))
                        && Arrays.equals(entries.get(1).getData(), "entry1".getBytes(Encoding))));
        entries.forEach(Entry::release);

        // 3. Fail on reading non-existing position
        PositionImpl invalidPosition = new PositionImpl(100, 100);
        positions.add(invalidPosition);

        try {
            c1.replayEntries(positions);
            fail("Should fail");
        } catch (ManagedLedgerException e) {
            // ok
        }
        positions.remove(invalidPosition);

        // 4. Fail to attempt to read mark-deleted position (p1)
        c1.markDelete(p2);

        try {
            // as mark-delete is at position: p2 it should read entry : p3
            assertEquals(1, c1.replayEntries(positions).size());
        } catch (ManagedLedgerException e) {
            fail("Should have not failed");
        }
    }

    @Test(timeOut = 20000)
    void testGetLastIndividualDeletedRange() throws Exception {
        ManagedLedger ledger = factory.open("test_last_individual_deleted");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        PositionImpl markDeletedPosition = (PositionImpl) c1.getMarkDeletedPosition();
        for(int i = 0; i < 10; i++) {
            ledger.addEntry(("entry" + i).getBytes(Encoding));
        }
        PositionImpl p1 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 1);
        PositionImpl p2 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 2);
        PositionImpl p3 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 5);
        PositionImpl p4 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 6);

        c1.delete(Lists.newArrayList(p1, p2, p3, p4));

        assertEquals(c1.getLastIndividualDeletedRange(), Range.openClosed(PositionImpl.get(p3.getLedgerId(),
                p3.getEntryId() - 1), p4));

        PositionImpl p5 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 8);
        c1.delete(p5);

        assertEquals(c1.getLastIndividualDeletedRange(), Range.openClosed(PositionImpl.get(p5.getLedgerId(),
                p5.getEntryId() - 1), p5));

    }

    @Test(timeOut = 20000)
    void testTrimDeletedEntries() throws ManagedLedgerException, InterruptedException {
        ManagedLedger ledger = factory.open("my_test_ledger");

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        PositionImpl markDeletedPosition = (PositionImpl) c1.getMarkDeletedPosition();
        for(int i = 0; i < 10; i++) {
            ledger.addEntry(("entry" + i).getBytes(Encoding));
        }
        PositionImpl p1 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 1);
        PositionImpl p2 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 2);
        PositionImpl p3 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 5);
        PositionImpl p4 = PositionImpl.get(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 6);

        c1.delete(Lists.newArrayList(p1, p2, p3, p4));

        EntryImpl entry1 = EntryImpl.create(p1, ByteBufAllocator.DEFAULT.buffer(0));
        EntryImpl entry2 = EntryImpl.create(p2, ByteBufAllocator.DEFAULT.buffer(0));
        EntryImpl entry3 = EntryImpl.create(p3, ByteBufAllocator.DEFAULT.buffer(0));
        EntryImpl entry4 = EntryImpl.create(p4, ByteBufAllocator.DEFAULT.buffer(0));
        EntryImpl entry5 = EntryImpl.create(markDeletedPosition.getLedgerId() , markDeletedPosition.getEntryId() + 7,
                ByteBufAllocator.DEFAULT.buffer(0));
        List<Entry> entries = Lists.newArrayList(entry1, entry2, entry3, entry4, entry5);
        c1.trimDeletedEntries(entries);
        assertEquals(entries.size(), 1);
        assertEquals(entries.get(0).getPosition(), PositionImpl.get(markDeletedPosition.getLedgerId() ,
                markDeletedPosition.getEntryId() + 7));
    }

    @Test(timeOut = 20000)
    void outOfOrderAcks() throws Exception {
        ManagedLedger ledger = factory.open("outOfOrderAcks");
        ManagedCursor c1 = ledger.openCursor("c1");

        int N = 10;

        List<Position> positions = new ArrayList<>();
        for (int i = 0; i < N; i++) {
            positions.add(ledger.addEntry("entry".getBytes()));
        }

        assertEquals(c1.getNumberOfEntriesInBacklog(false), N);

        c1.delete(positions.get(3));
        assertEquals(c1.getNumberOfEntriesInBacklog(false), N - 1);

        c1.delete(positions.get(2));
        assertEquals(c1.getNumberOfEntriesInBacklog(false), N - 2);

        c1.delete(positions.get(1));
        assertEquals(c1.getNumberOfEntriesInBacklog(false), N - 3);

        c1.delete(positions.get(0));
        assertEquals(c1.getNumberOfEntriesInBacklog(false), N - 4);
    }

    @Test(timeOut = 20000)
    void randomOrderAcks() throws Exception {
        ManagedLedger ledger = factory.open("outOfOrderAcks");
        ManagedCursor c1 = ledger.openCursor("c1");

        int N = 10;

        List<Position> positions = new ArrayList<>();
        for (int i = 0; i < N; i++) {
            positions.add(ledger.addEntry("entry".getBytes()));
        }

        assertEquals(c1.getNumberOfEntriesInBacklog(false), N);

        // Randomize the ack sequence
        Collections.shuffle(positions);

        int toDelete = N;
        for (Position p : positions) {
            assertEquals(c1.getNumberOfEntriesInBacklog(false), toDelete);
            c1.delete(p);
            --toDelete;
            assertEquals(c1.getNumberOfEntriesInBacklog(false), toDelete);
        }
    }

    @Test(timeOut = 20000)
    void testGetEntryAfterN() throws Exception {
        ManagedLedger ledger = factory.open("testGetEntryAfterN");

        ManagedCursor c1 = ledger.openCursor("c1");

        Position pos1 = ledger.addEntry("msg1".getBytes());
        Position pos2 = ledger.addEntry("msg2".getBytes());
        Position pos3 = ledger.addEntry("msg3".getBytes());
        Position pos4 = ledger.addEntry("msg4".getBytes());
        Position pos5 = ledger.addEntry("msg5".getBytes());

        List<Entry> entries = c1.readEntries(4);
        entries.forEach(e -> e.release());
        long currentLedger = ((PositionImpl) c1.getMarkDeletedPosition()).getLedgerId();

        // check if the first message is returned for '0'
        Entry e = c1.getNthEntry(1, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg1".getBytes());

        // check that if we call get entry for the same position twice, it returns the same entry
        e = c1.getNthEntry(1, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg1".getBytes());

        // check for a position 'n' after md position
        e = c1.getNthEntry(3, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg3".getBytes());

        // check for the last position
        e = c1.getNthEntry(5, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg5".getBytes());

        // check for a position outside the limits of the number of entries that exists, it should return null
        e = c1.getNthEntry(10, IndividualDeletedEntries.Exclude);
        assertNull(e);

        // check that the mark delete and read positions have not been updated after all the previous operations
        assertEquals(c1.getMarkDeletedPosition(), new PositionImpl(currentLedger, -1));
        assertEquals(c1.getReadPosition(), new PositionImpl(currentLedger, 4));

        c1.markDelete(pos4);
        assertEquals(c1.getMarkDeletedPosition(), pos4);
        e = c1.getNthEntry(1, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg5".getBytes());

        c1.readEntries(1);
        c1.markDelete(pos5);

        e = c1.getNthEntry(1, IndividualDeletedEntries.Exclude);
        assertNull(e);
    }

    @Test(timeOut = 20000)
    void testGetEntryAfterNWithIndividualDeletedMessages() throws Exception {
        ManagedLedger ledger = factory.open("testGetEnteryAfterNWithIndividualDeletedMessages");

        ManagedCursor c1 = ledger.openCursor("c1");

        Position pos1 = ledger.addEntry("msg1".getBytes());
        Position pos2 = ledger.addEntry("msg2".getBytes());
        Position pos3 = ledger.addEntry("msg3".getBytes());
        Position pos4 = ledger.addEntry("msg4".getBytes());
        Position pos5 = ledger.addEntry("msg5".getBytes());

        c1.delete(pos3);
        c1.delete(pos4);

        Entry e = c1.getNthEntry(3, IndividualDeletedEntries.Exclude);
        assertEquals(e.getDataAndRelease(), "msg5".getBytes());

        e = c1.getNthEntry(3, IndividualDeletedEntries.Include);
        assertEquals(e.getDataAndRelease(), "msg3".getBytes());
    }

    @Test(timeOut = 20000)
    void cancelReadOperation() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));

        ManagedCursor c1 = ledger.openCursor("c1");

        // No read request so far
        assertFalse(c1.cancelPendingReadRequest());

        CountDownLatch counter = new CountDownLatch(1);

        c1.asyncReadEntriesOrWait(1, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                counter.countDown();
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                counter.countDown();
            }
        }, null, PositionImpl.latest);

        assertTrue(c1.cancelPendingReadRequest());

        CountDownLatch counter2 = new CountDownLatch(1);

        c1.asyncReadEntriesOrWait(1, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                counter2.countDown();
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                counter2.countDown();
            }
        }, null, PositionImpl.latest);

        ledger.addEntry("entry-1".getBytes(Encoding));

        Thread.sleep(100);

        // Read operation should have already been completed
        assertFalse(c1.cancelPendingReadRequest());

        counter2.await();
    }

    @Test(timeOut = 20000)
    public void testReopenMultipleTimes() throws Exception {
        ManagedLedger ledger = factory.open("testReopenMultipleTimes");
        ManagedCursor c1 = ledger.openCursor("c1");

        Position mdPosition = c1.getMarkDeletedPosition();

        c1.close();
        ledger.close();

        ledger = factory.open("testReopenMultipleTimes");
        c1 = ledger.openCursor("c1");

        // since the empty data ledger will be deleted, the cursor position should also be updated
        assertNotEquals(c1.getMarkDeletedPosition(), mdPosition);

        c1.close();
        ledger.close();

        ledger = factory.open("testReopenMultipleTimes");
        c1 = ledger.openCursor("c1");
    }

    @Test(timeOut = 20000)
    public void testOutOfOrderDeletePersistenceWithClose() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());

        ManagedCursor c1 = ledger.openCursor("c1");
        List<Position> addedPositions = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            Position p = ledger.addEntry(("dummy-entry-" + i).getBytes(Encoding));
            addedPositions.add(p);
        }

        // Acknowledge few messages leaving holes
        c1.delete(addedPositions.get(2));
        c1.delete(addedPositions.get(5));
        c1.delete(addedPositions.get(7));
        c1.delete(addedPositions.get(8));
        c1.delete(addedPositions.get(9));

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 20 - 5);

        ledger.close();
        factory.shutdown();

        // Re-Open
        factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());
        c1 = ledger.openCursor("c1");
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 20 - 5);

        List<Entry> entries = c1.readEntries(20);
        assertEquals(entries.size(), 20 - 5);

        List<String> entriesStr = entries.stream().map(e -> new String(e.getDataAndRelease(), Encoding))
                .collect(Collectors.toList());
        assertEquals(entriesStr.get(0), "dummy-entry-0");
        assertEquals(entriesStr.get(1), "dummy-entry-1");
        // Entry-2 was deleted
        assertEquals(entriesStr.get(2), "dummy-entry-3");
        assertEquals(entriesStr.get(3), "dummy-entry-4");
        // Entry-6 was deleted
        assertEquals(entriesStr.get(4), "dummy-entry-6");

        assertFalse(c1.hasMoreEntries());
    }

    @Test(timeOut = 20000)
    public void testOutOfOrderDeletePersistenceAfterCrash() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());

        ManagedCursor c1 = ledger.openCursor("c1");
        List<Position> addedPositions = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            Position p = ledger.addEntry(("dummy-entry-" + i).getBytes(Encoding));
            addedPositions.add(p);
        }

        // Acknowledge few messages leaving holes
        c1.delete(addedPositions.get(2));
        c1.delete(addedPositions.get(5));
        c1.delete(addedPositions.get(7));
        c1.delete(addedPositions.get(8));
        c1.delete(addedPositions.get(9));

        assertEquals(c1.getNumberOfEntriesInBacklog(false), 20 - 5);

        // Re-Open
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig());
        c1 = ledger.openCursor("c1");
        assertEquals(c1.getNumberOfEntriesInBacklog(false), 20 - 5);

        List<Entry> entries = c1.readEntries(20);
        assertEquals(entries.size(), 20 - 5);

        List<String> entriesStr = entries.stream().map(e -> new String(e.getDataAndRelease(), Encoding))
                .collect(Collectors.toList());
        assertEquals(entriesStr.get(0), "dummy-entry-0");
        assertEquals(entriesStr.get(1), "dummy-entry-1");
        // Entry-2 was deleted
        assertEquals(entriesStr.get(2), "dummy-entry-3");
        assertEquals(entriesStr.get(3), "dummy-entry-4");
        // Entry-6 was deleted
        assertEquals(entriesStr.get(4), "dummy-entry-6");

        assertFalse(c1.hasMoreEntries());
    }

    /**
     * <pre>
     * Verifies that {@link ManagedCursorImpl#createNewMetadataLedger()} cleans up orphan ledgers if fails to switch new
     * ledger
     * </pre>
     * @throws Exception
     */
    @Test(timeOut=5000)
    public void testLeakFailedLedgerOfManageCursor() throws Exception {

        ManagedLedgerConfig mlConfig = new ManagedLedgerConfig();
        ManagedLedger ledger = factory.open("my_test_ledger", mlConfig);

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        CountDownLatch latch = new CountDownLatch(1);
        c1.createNewMetadataLedger(new VoidCallback() {
            @Override
            public void operationComplete() {
                latch.countDown();
            }

            @Override
            public void operationFailed(ManagedLedgerException exception) {
                latch.countDown();
            }
        });

        // update cursor-info with data which makes bad-version for existing managed-cursor
        String path = "/managed-ledgers/my_test_ledger/c1";
        metadataStore.put(path, "".getBytes(), Optional.empty()).join();

        // try to create ledger again which will fail because managedCursorInfo znode is already updated with different
        // version so, this call will fail with BadVersionException
        CountDownLatch latch2 = new CountDownLatch(1);
        // create ledger will create ledgerId = 6
        long ledgerId = 6;
        c1.createNewMetadataLedger(new VoidCallback() {
            @Override
            public void operationComplete() {
                latch2.countDown();
            }

            @Override
            public void operationFailed(ManagedLedgerException exception) {
                latch2.countDown();
            }
        });

        // Wait until operation is completed and the failed ledger should have been deleted
        latch2.await();

        try {
            bkc.openLedgerNoRecovery(ledgerId, DigestType.fromApiDigestType(mlConfig.getDigestType()),
                                     mlConfig.getPassword());
            fail("ledger should have deleted due to update-cursor failure");
        } catch (BKException e) {
            // ok
        }
    }

    /**
     * Verifies cursor persists individually unack range into cursor-ledger if range count is higher than
     * MaxUnackedRangesToPersistInZk
     *
     * @throws Exception
     */
    @Test(timeOut = 20000)
    public void testOutOfOrderDeletePersistenceIntoLedgerWithClose() throws Exception {

        final int totalAddEntries = 100;
        String ledgerName = "my_test_ledger";
        String cursorName = "c1";
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        // metaStore is allowed to store only up to 10 deleted entries range
        managedLedgerConfig.setMaxUnackedRangesToPersistInZk(10);
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerName, managedLedgerConfig);

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(cursorName);

        List<Position> addedPositions = new ArrayList<>();
        for (int i = 0; i < totalAddEntries; i++) {
            Position p = ledger.addEntry(("dummy-entry-" + i).getBytes(Encoding));
            addedPositions.add(p);
            if (i % 2 == 0) {
                // Acknowledge alternative message to create totalEntries/2 holes
                c1.delete(addedPositions.get(i));
            }
        }

        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalAddEntries / 2);

        // Close ledger to persist individual-deleted positions into cursor-ledger
        ledger.close();

        // verify cursor-ledgerId is updated properly into cursor-metaStore
        CountDownLatch cursorLedgerLatch = new CountDownLatch(1);
        AtomicLong cursorLedgerId = new AtomicLong(0);
        ledger.getStore().asyncGetCursorInfo(ledger.getName(), cursorName, new MetaStoreCallback<ManagedCursorInfo>() {
            @Override
            public void operationComplete(ManagedCursorInfo result, Stat stat) {
                cursorLedgerId.set(result.getCursorsLedgerId());
                cursorLedgerLatch.countDown();
            }

            @Override
            public void operationFailed(MetaStoreException e) {
                cursorLedgerLatch.countDown();
            }
        });
        cursorLedgerLatch.await();
        assertEquals(cursorLedgerId.get(), c1.getCursorLedger());

        // verify cursor-ledger's last entry has individual-deleted positions
        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicInteger individualDeletedMessagesCount = new AtomicInteger(0);
        bkc.asyncOpenLedger(c1.getCursorLedger(), DigestType.CRC32C, "".getBytes(), (rc, lh, ctx) -> {
            if (rc == BKException.Code.OK) {
                long lastEntry = lh.getLastAddConfirmed();
                lh.asyncReadEntries(lastEntry, lastEntry, (rc1, lh1, seq, ctx1) -> {
                    try {
                        LedgerEntry entry = seq.nextElement();
                        PositionInfo positionInfo;
                        positionInfo = PositionInfo.parseFrom(entry.getEntry());
                        individualDeletedMessagesCount.set(positionInfo.getIndividualDeletedMessagesCount());
                    } catch (Exception e) {
                    }
                    latch.countDown();
                }, null);
            } else {
                latch.countDown();
            }
        }, null);

        latch.await();
        assertEquals(individualDeletedMessagesCount.get(), totalAddEntries / 2 - 1);

        // Re-Open
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = (ManagedLedgerImpl) factory2.open(ledgerName, managedLedgerConfig);
        c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        // verify cursor has been recovered
        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalAddEntries / 2);

        // try to read entries which should only read non-deleted positions
        List<Entry> entries = c1.readEntries(totalAddEntries);
        assertEquals(entries.size(), totalAddEntries / 2);
    }

    /**
     * Close Cursor without MaxUnackedRangesToPersistInZK: It should store individually unack range into Zk
     *
     * @throws Exception
     */
    @Test(timeOut = 20000)
    public void testOutOfOrderDeletePersistenceIntoZkWithClose() throws Exception {
        final int totalAddEntries = 100;
        String ledgerName = "my_test_ledger_zk";
        String cursorName = "c1";
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerName, managedLedgerConfig);

        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(cursorName);

        List<Position> addedPositions = new ArrayList<>();
        for (int i = 0; i < totalAddEntries; i++) {
            Position p = ledger.addEntry(("dummy-entry-" + i).getBytes(Encoding));
            addedPositions.add(p);
            if (i % 2 == 0) {
                // Acknowledge alternative message to create totalEntries/2 holes
                c1.delete(addedPositions.get(i));
            }
        }

        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalAddEntries / 2);

        // Close ledger to persist individual-deleted positions into cursor-ledger
        ledger.close();

        // verify cursor-ledgerId is updated as -1 into cursor-metaStore
        CountDownLatch latch = new CountDownLatch(1);
        AtomicInteger individualDeletedMessagesCount = new AtomicInteger(0);
        ledger.getStore().asyncGetCursorInfo(ledger.getName(), cursorName, new MetaStoreCallback<ManagedCursorInfo>() {
            @Override
            public void operationComplete(ManagedCursorInfo result, Stat stat) {
                individualDeletedMessagesCount.set(result.getIndividualDeletedMessagesCount());
                latch.countDown();
            }

            @Override
            public void operationFailed(MetaStoreException e) {
                latch.countDown();
            }
        });
        latch.await();
        assertEquals(individualDeletedMessagesCount.get(), totalAddEntries / 2 - 1);

        // Re-Open
        @Cleanup("shutdown")
        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ledger = (ManagedLedgerImpl) factory2.open(ledgerName, managedLedgerConfig);
        c1 = (ManagedCursorImpl) ledger.openCursor(cursorName);
        // verify cursor has been recovered
        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalAddEntries / 2);

        // try to read entries which should only read non-deleted positions
        List<Entry> entries = c1.readEntries(totalAddEntries);
        assertEquals(entries.size(), totalAddEntries / 2);
    }

    @Test
    public void testInvalidMarkDelete() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());

        ManagedCursor cursor = ledger.openCursor("c1");
        Position readPosition = cursor.getReadPosition();
        Position markDeletePosition = cursor.getMarkDeletedPosition();

        List<Position> addedPositions = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            Position p = ledger.addEntry(("dummy-entry-" + i).getBytes(Encoding));
            addedPositions.add(p);
        }

        // validate: cursor.asyncMarkDelete(..)
        CountDownLatch markDeleteCallbackLatch = new CountDownLatch(1);
        Position position = PositionImpl.get(100, 100);
        AtomicBoolean markDeleteCallFailed = new AtomicBoolean(false);
        cursor.asyncMarkDelete(position, new MarkDeleteCallback() {
            @Override
            public void markDeleteComplete(Object ctx) {
                markDeleteCallbackLatch.countDown();
            }

            @Override
            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                markDeleteCallFailed.set(true);
                markDeleteCallbackLatch.countDown();
            }
        }, null);
        markDeleteCallbackLatch.await();
        assertEquals(readPosition, cursor.getReadPosition());
        assertEquals(markDeletePosition, cursor.getMarkDeletedPosition());

        // validate : cursor.asyncDelete(..)
        CountDownLatch deleteCallbackLatch = new CountDownLatch(1);
        markDeleteCallFailed.set(false);
        cursor.asyncDelete(position, new DeleteCallback() {
            @Override
            public void deleteComplete(Object ctx) {
                deleteCallbackLatch.countDown();
            }

            @Override
            public void deleteFailed(ManagedLedgerException exception, Object ctx) {
                markDeleteCallFailed.set(true);
                deleteCallbackLatch.countDown();
            }
        }, null);

        deleteCallbackLatch.await();
        assertEquals(readPosition, cursor.getReadPosition());
        assertEquals(markDeletePosition, cursor.getMarkDeletedPosition());
    }

    @Test
    public void testEstimatedUnackedSize() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setMaxEntriesPerLedger(10);
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());

        ManagedCursor cursor = ledger.openCursor("c1");

        byte[] entryData = new byte[5];

        // write 15 entries, saving position of 5th
        for (int i = 0; i < 4; i++) { ledger.addEntry(entryData); }
        Position deleteAt = ledger.addEntry(entryData);
        for (int i = 0; i < 10; i++) { ledger.addEntry(entryData); }

        assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 15 * entryData.length);

        cursor.markDelete(deleteAt);

        // it's not an estimate if all entries are the same size
        assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 10 * entryData.length);
    }

    @Test(timeOut = 20000)
    public void testRecoverCursorAheadOfLastPosition() throws Exception {
        final String mlName = "my_test_ledger";
        final PositionImpl lastPosition = new PositionImpl(1L, 10L);
        final PositionImpl nextPosition = new PositionImpl(3L, -1L);

        final String cursorName = "my_test_cursor";
        final long cursorsLedgerId = -1L;
        final long markDeleteLedgerId = 2L;
        final long markDeleteEntryId = -1L;

        MetaStore mockMetaStore = mock(MetaStore.class);
        doAnswer(new Answer<Object>() {
            public Object answer(InvocationOnMock invocation) {
                ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(cursorsLedgerId)
                        .setMarkDeleteLedgerId(markDeleteLedgerId).setMarkDeleteEntryId(markDeleteEntryId)
                        .setLastActive(0L).build();
                Stat stat = mock(Stat.class);
                MetaStoreCallback<ManagedCursorInfo> callback = (MetaStoreCallback<ManagedCursorInfo>) invocation
                        .getArguments()[2];
                callback.operationComplete(info, stat);
                return null;
            }
        }).when(mockMetaStore).asyncGetCursorInfo(eq(mlName), eq(cursorName), any(MetaStoreCallback.class));

        ManagedLedgerImpl ml = mock(ManagedLedgerImpl.class);
        when(ml.getName()).thenReturn(mlName);
        when(ml.getStore()).thenReturn(mockMetaStore);
        when(ml.getLastPosition()).thenReturn(lastPosition);
        when(ml.getNextValidLedger(markDeleteLedgerId)).thenReturn(3L);
        when(ml.getNextValidPosition(lastPosition)).thenReturn(nextPosition);
        when(ml.ledgerExists(markDeleteLedgerId)).thenReturn(false);

        BookKeeper mockBookKeeper = mock(BookKeeper.class);
        final ManagedCursorImpl cursor = new ManagedCursorImpl(mockBookKeeper, new ManagedLedgerConfig(), ml,
                cursorName);

        cursor.recover(new VoidCallback() {
            @Override
            public void operationComplete() {
                assertEquals(cursor.getMarkDeletedPosition(), lastPosition);
                assertEquals(cursor.getReadPosition(), nextPosition);
                assertEquals(cursor.getNumberOfEntries(), 0L);
            }

            @Override
            public void operationFailed(ManagedLedgerException exception) {
                fail("Cursor recovery should not fail");
            }
        });
    }

    @Test
    void testAlwaysInactive() throws Exception {
        ManagedLedger ml = factory.open("testAlwaysInactive");
        ManagedCursor cursor = ml.openCursor("c1");

        assertTrue(cursor.isActive());

        cursor.setAlwaysInactive();

        assertFalse(cursor.isActive());

        cursor.setActive();
        assertFalse(cursor.isActive());
    }

    @Test
    void testNonDurableCursorActive() throws Exception {
        ManagedLedger ml = factory.open("testInactive");
        ManagedCursor cursor = ml.newNonDurableCursor(PositionImpl.latest, "c1");

        assertTrue(cursor.isActive());

        cursor.setInactive();
        assertFalse(cursor.isActive());
    }

    @Test
    public void deleteMessagesCheckhMarkDelete() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger");
        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
        final int totalEntries = 1000;
        final Position[] positions = new Position[totalEntries];
        for (int i = 0; i < totalEntries; i++) {
            // add entry
            positions[i] = ledger.addEntry(("entry-" + i).getBytes(Encoding));
        }
        assertEquals(c1.getNumberOfEntries(), totalEntries);
        int totalDeletedMessages = 0;
        for (int i = 0; i < totalEntries; i++) {
            // delete entry
            if ((i % 3) == 0) {
                c1.delete(positions[i]);
                totalDeletedMessages += 1;
            }
        }
        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalEntries - totalDeletedMessages);
        assertEquals(c1.getNumberOfEntries(), totalEntries - totalDeletedMessages);
        assertEquals(c1.getMarkDeletedPosition(), positions[0]);
        assertEquals(c1.getReadPosition(), positions[1]);

        // delete 1/2 of the messags
        for (int i = 0; i < totalEntries / 2; i++) {
            // delete entry
            if ((i % 3) != 0) {
                c1.delete(positions[i]);
                totalDeletedMessages += 1;
            }
        }
        int markDelete = totalEntries / 2 - 1;
        assertEquals(c1.getNumberOfEntriesInBacklog(false), totalEntries - totalDeletedMessages);
        assertEquals(c1.getNumberOfEntries(), totalEntries - totalDeletedMessages);
        assertEquals(c1.getMarkDeletedPosition(), positions[markDelete]);
        assertEquals(c1.getReadPosition(), positions[markDelete + 1]);
    }

    @Test
    public void testBatchIndexDelete() throws ManagedLedgerException, InterruptedException {
        ManagedLedger ledger = factory.open("test_batch_index_delete");
        ManagedCursor cursor = ledger.openCursor("c1");

        final int totalEntries = 100;
        final Position[] positions = new Position[totalEntries];
        for (int i = 0; i < totalEntries; i++) {
            // add entry
            positions[i] = ledger.addEntry(("entry-" + i).getBytes(Encoding));
        }
        assertEquals(cursor.getNumberOfEntries(), totalEntries);
        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(2).setEnd(4)));
        List<IntRange> deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[0]), 10);
        Assert.assertEquals(1, deletedIndexes.size());
        Assert.assertEquals(2, deletedIndexes.get(0).getStart());
        Assert.assertEquals(4, deletedIndexes.get(0).getEnd());

        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(3).setEnd(8)));
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[0]), 10);
        Assert.assertEquals(1, deletedIndexes.size());
        Assert.assertEquals(2, deletedIndexes.get(0).getStart());
        Assert.assertEquals(8, deletedIndexes.get(0).getEnd());

        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(0)));
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[0]), 10);
        Assert.assertEquals(2, deletedIndexes.size());
        Assert.assertEquals(0, deletedIndexes.get(0).getStart());
        Assert.assertEquals(0, deletedIndexes.get(0).getEnd());
        Assert.assertEquals(2, deletedIndexes.get(1).getStart());
        Assert.assertEquals(8, deletedIndexes.get(1).getEnd());

        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(1).setEnd(1)));
        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(9).setEnd(9)));
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[0]), 10);
        Assert.assertNull(deletedIndexes);
        Assert.assertEquals(positions[0], cursor.getMarkDeletedPosition());

        deleteBatchIndex(cursor, positions[1], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(5)));
        cursor.delete(positions[1]);
        deleteBatchIndex(cursor, positions[1], 10, Lists.newArrayList(new IntRange().setStart(6).setEnd(8)));
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[1]), 10);
        Assert.assertNull(deletedIndexes);

        deleteBatchIndex(cursor, positions[2], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(5)));
        cursor.markDelete(positions[3]);
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[2]), 10);
        Assert.assertNull(deletedIndexes);

        deleteBatchIndex(cursor, positions[3], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(5)));
        cursor.resetCursor(positions[0]);
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[3]), 10);
        Assert.assertNull(deletedIndexes);
    }

    @Test
    public void testBatchIndexesDeletionPersistAndRecover() throws ManagedLedgerException, InterruptedException {
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        // Make sure the cursor metadata updated by the cursor ledger ID.
        managedLedgerConfig.setMaxUnackedRangesToPersistInZk(-1);
        ManagedLedger ledger = factory.open("test_batch_indexes_deletion_persistent", managedLedgerConfig);
        ManagedCursor cursor = ledger.openCursor("c1");

        final int totalEntries = 100;
        final Position[] positions = new Position[totalEntries];
        for (int i = 0; i < totalEntries; i++) {
            // add entry
            positions[i] = ledger.addEntry(("entry-" + i).getBytes(Encoding));
        }
        assertEquals(cursor.getNumberOfEntries(), totalEntries);
        deleteBatchIndex(cursor, positions[6], 10, Lists.newArrayList(new IntRange().setStart(1).setEnd(3)));
        deleteBatchIndex(cursor, positions[5], 10, Lists.newArrayList(new IntRange().setStart(3).setEnd(6)));
        deleteBatchIndex(cursor, positions[0], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));
        deleteBatchIndex(cursor, positions[1], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));
        deleteBatchIndex(cursor, positions[2], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));
        deleteBatchIndex(cursor, positions[3], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));
        deleteBatchIndex(cursor, positions[4], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));

        cursor.close();
        ledger.close();
        ledger = factory.open("test_batch_indexes_deletion_persistent", managedLedgerConfig);
        cursor = ledger.openCursor("c1");

        List<IntRange> deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[5]), 10);
        Assert.assertEquals(deletedIndexes.size(), 1);
        Assert.assertEquals(deletedIndexes.get(0).getStart(), 3);
        Assert.assertEquals(deletedIndexes.get(0).getEnd(), 6);
        Assert.assertEquals(cursor.getMarkDeletedPosition(), positions[4]);
        deleteBatchIndex(cursor, positions[5], 10, Lists.newArrayList(new IntRange().setStart(0).setEnd(9)));
        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[5]), 10);
        Assert.assertNull(deletedIndexes);
        Assert.assertEquals(cursor.getMarkDeletedPosition(), positions[5]);

        deletedIndexes = getAckedIndexRange(cursor.getDeletedBatchIndexesAsLongArray((PositionImpl) positions[6]), 10);
        Assert.assertEquals(deletedIndexes.size(), 1);
        Assert.assertEquals(deletedIndexes.get(0).getStart(), 1);
        Assert.assertEquals(deletedIndexes.get(0).getEnd(), 3);
    }

    private void deleteBatchIndex(ManagedCursor cursor, Position position, int batchSize,
                                  List<IntRange> deleteIndexes) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        PositionImpl pos = (PositionImpl) position;
        BitSet bitSet = new BitSet(batchSize);
        bitSet.set(0, batchSize);
        deleteIndexes.forEach(intRange -> {
            bitSet.clear(intRange.getStart(), intRange.getEnd() + 1);
        });
        pos.ackSet = bitSet.toLongArray();

        cursor.asyncDelete(pos,
            new DeleteCallback() {
                @Override
                public void deleteComplete(Object ctx) {
                    latch.countDown();
                }

                @Override
                public void deleteFailed(ManagedLedgerException exception, Object ctx) {
                    latch.countDown();
                }
            }, null);
        latch.await();
        pos.ackSet = null;
    }

    private List<IntRange> getAckedIndexRange(long[] bitSetLongArray, int batchSize) {
        if (bitSetLongArray == null) {
            return null;
        }
        List<IntRange> result = new ArrayList<>();
        BitSet bitSet = BitSet.valueOf(bitSetLongArray);
        int nextClearBit = bitSet.nextClearBit(0);
        while (nextClearBit != -1 && nextClearBit <= batchSize) {
            int nextSetBit = bitSet.nextSetBit(nextClearBit);
            if (nextSetBit == -1) {
                break;
            }
            result.add(new IntRange().setStart(nextClearBit).setEnd(nextSetBit - 1));
            nextClearBit = bitSet.nextClearBit(nextSetBit);
        }
        return result;
    }

    @Test
    public void testReadEntriesOrWaitWithMaxSize() throws Exception {
        ManagedLedger ledger = factory.open("testReadEntriesOrWaitWithMaxSize");
        ManagedCursor c = ledger.openCursor("c");

        for (int i = 0; i < 20; i++) {
            ledger.addEntry(new byte[1024]);
        }

        // First time, since we don't have info, we'll get 1 single entry
        List<Entry> entries = c.readEntriesOrWait(10, 3 * 1024);
        assertEquals(entries.size(), 1);
        entries.forEach(e -> e.release());

        // We should only return 3 entries, based on the max size
        entries = c.readEntriesOrWait(10, 3 * 1024);
        assertEquals(entries.size(), 3);
        entries.forEach(e -> e.release());

        // If maxSize is < avg, we should get 1 entry
        entries = c.readEntriesOrWait(10, 5);
        assertEquals(entries.size(), 1);
        entries.forEach(e -> e.release());
    }

    @Test
    public void testReadEntriesOrWaitWithMaxPosition() throws Exception {
        int readMaxNumber = 10;
        int sendNumber = 20;
        ManagedLedger ledger = factory.open("testReadEntriesOrWaitWithMaxPosition");
        ManagedCursor c = ledger.openCursor("c");
        Position position = PositionImpl.earliest;
        Position maxCanReadPosition = PositionImpl.earliest;
        for (int i = 0; i < sendNumber; i++) {
            if (i == readMaxNumber - 1) {
                position = ledger.addEntry(new byte[1024]);
            } else if (i == sendNumber - 1) {
                maxCanReadPosition = ledger.addEntry(new byte[1024]);
            } else {
                ledger.addEntry(new byte[1024]);
            }

        }
        CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
        c.asyncReadEntriesOrWait(sendNumber, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                completableFuture.complete(entries.size());
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                completableFuture.completeExceptionally(exception);
            }
        }, null, (PositionImpl) position);

        int number = completableFuture.get();
        assertEquals(number, readMaxNumber);

        c.asyncReadEntriesOrWait(sendNumber, new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                completableFuture.complete(entries.size());
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                completableFuture.completeExceptionally(exception);
            }
        }, null, (PositionImpl) maxCanReadPosition);

        assertEquals(number, sendNumber - readMaxNumber);

    }

    @Test
    public void testFlushCursorAfterInactivity() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setThrottleMarkDelete(1.0);

        ManagedLedgerFactoryConfig factoryConfig = new ManagedLedgerFactoryConfig();
        factoryConfig.setCursorPositionFlushSeconds(1);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory1 = new ManagedLedgerFactoryImpl(metadataStore, bkc, factoryConfig);
        ManagedLedger ledger1 = factory1.open("testFlushCursorAfterInactivity", config);
        ManagedCursor c1 = ledger1.openCursor("c");
        List<Position> positions = new ArrayList<Position>();

        for (int i = 0; i < 20; i++) {
            positions.add(ledger1.addEntry(new byte[1024]));
        }

        CountDownLatch latch = new CountDownLatch(positions.size());

        positions.forEach(p -> c1.asyncMarkDelete(p, new MarkDeleteCallback() {
            @Override
            public void markDeleteComplete(Object ctx) {
                latch.countDown();
            }

            @Override
            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                throw new RuntimeException(exception);
            }
        }, null));

        latch.await();

        assertEquals(c1.getMarkDeletedPosition(), positions.get(positions.size() - 1));

        Awaitility.await()
                // Give chance to the flush to be automatically triggered.
                // NOTE: this can't be set too low, or it causes issues with ZK thread pool rejecting
                .pollDelay(Duration.ofMillis(2000))
                .untilAsserted(() -> {
                    // Abruptly re-open the managed ledger without graceful close
                    @Cleanup("shutdown")
                    ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
                    ManagedLedger ledger2 = factory2.open("testFlushCursorAfterInactivity", config);
                    ManagedCursor c2 = ledger2.openCursor("c");

                    assertEquals(c2.getMarkDeletedPosition(), positions.get(positions.size() - 1));
                });
    }

    @Test
    public void testFlushCursorAfterIndividualDeleteInactivity() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setThrottleMarkDelete(1.0);

        ManagedLedgerFactoryConfig factoryConfig = new ManagedLedgerFactoryConfig();
        factoryConfig.setCursorPositionFlushSeconds(1);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory1 = new ManagedLedgerFactoryImpl(metadataStore, bkc, factoryConfig);
        ManagedLedger ledger1 = factory1.open("testFlushCursorAfterIndDelInactivity", config);
        ManagedCursor c1 = ledger1.openCursor("c");
        List<Position> positions = new ArrayList<Position>();

        for (int i = 0; i < 20; i++) {
            positions.add(ledger1.addEntry(new byte[1024]));
        }

        CountDownLatch latch = new CountDownLatch(positions.size());

        positions.forEach(p -> c1.asyncDelete(p, new DeleteCallback() {
            @Override
            public void deleteComplete(Object ctx) {
                latch.countDown();
            }

            @Override
            public void deleteFailed(ManagedLedgerException exception, Object ctx) {
                throw new RuntimeException(exception);
            }
        }, null));

        latch.await();

        assertEquals(c1.getMarkDeletedPosition(), positions.get(positions.size() - 1));

        // reopen the cursor and we should see entries not be flushed
        @Cleanup("shutdown")
        ManagedLedgerFactory dirtyFactory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
        ManagedLedger ledgerDirty = dirtyFactory.open("testFlushCursorAfterIndDelInactivity", config);
        ManagedCursor dirtyCursor = ledgerDirty.openCursor("c");

        assertNotEquals(dirtyCursor.getMarkDeletedPosition(), positions.get(positions.size() - 1));

        Awaitility.await()
                // Give chance to the flush to be automatically triggered.
                // NOTE: this can't be set too low, or it causes issues with ZK thread pool rejecting
                .pollDelay(Duration.ofMillis(2000))
                .untilAsserted(() -> {
                    // Abruptly re-open the managed ledger without graceful close
                    @Cleanup("shutdown")
                    ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
                    ManagedLedger ledger2 = factory2.open("testFlushCursorAfterIndDelInactivity", config);
                    ManagedCursor c2 = ledger2.openCursor("c");

                    assertEquals(c2.getMarkDeletedPosition(), positions.get(positions.size() - 1));
                });
    }


    @Test
    public void testFlushCursorAfterError() throws Exception {
        ManagedLedgerConfig config = new ManagedLedgerConfig();
        config.setThrottleMarkDelete(1.0);

        ManagedLedgerFactoryConfig factoryConfig = new ManagedLedgerFactoryConfig();
        factoryConfig.setCursorPositionFlushSeconds(1);

        @Cleanup("shutdown")
        ManagedLedgerFactory factory1 = new ManagedLedgerFactoryImpl(metadataStore, bkc, factoryConfig);
        ManagedLedger ledger1 = factory1.open("testFlushCursorAfterInactivity", config);
        ManagedCursor c1 = ledger1.openCursor("c");
        List<Position> positions = new ArrayList<>();

        for (int i = 0; i < 20; i++) {
            positions.add(ledger1.addEntry(new byte[1024]));
        }

        // Simulate BK write error
        bkc.failNow(BKException.Code.NotEnoughBookiesException);
        metadataStore.setAlwaysFail(new MetadataStoreException.BadVersionException(""));

        try {
            c1.markDelete(positions.get(positions.size() - 1));
            fail("should have failed");
        } catch (ManagedLedgerException e) {
            // Expected
        }

        metadataStore.unsetAlwaysFail();

        // In memory position is updated
        assertEquals(c1.getMarkDeletedPosition(), positions.get(positions.size() - 1));

        Awaitility.await()
                // Give chance to the flush to be automatically triggered.
                // NOTE: this can't be set too low, or it causes issues with ZK thread pool rejecting
                .pollDelay(Duration.ofMillis(2000))
                .untilAsserted(() -> {
                    // Abruptly re-open the managed ledger without graceful close
                    @Cleanup("shutdown")
                    ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
                    ManagedLedger ledger2 = factory2.open("testFlushCursorAfterInactivity", config);
                    ManagedCursor c2 = ledger2.openCursor("c");

                    assertEquals(c2.getMarkDeletedPosition(), positions.get(positions.size() - 1));
                });
    }

    @Test
    public void testConsistencyOfIndividualMessages() throws Exception {
        ManagedLedger ledger1 = factory.open("testConsistencyOfIndividualMessages");
        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger1.openCursor("c");

        PositionImpl p1 = (PositionImpl) ledger1.addEntry(new byte[1024]);
        c1.markDelete(p1);

        // Artificially add a position that is before the current mark-delete position
        LongPairRangeSet<PositionImpl> idm = c1.getIndividuallyDeletedMessagesSet();
        idm.addOpenClosed(p1.getLedgerId() - 1, 0, p1.getLedgerId() - 1, 10);

        List<Position> positions = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            positions.add(ledger1.addEntry(new byte[1024]));
        }

        for (int i = 0; i < 20; i++) {
            c1.delete(positions.get(i));
        }

        assertEquals(c1.getTotalNonContiguousDeletedMessagesRange(), 0);
        assertEquals(c1.getMarkDeletedPosition(), positions.get(positions.size() -1));
    }

    @Test
    public void testCursorCheckReadPositionChanged() throws Exception {
        ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());
        ManagedCursor c1 = ledger.openCursor("c1");

        // check empty ledger
        assertTrue(c1.checkAndUpdateReadPositionChanged());
        assertTrue(c1.checkAndUpdateReadPositionChanged());

        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));
        ledger.addEntry("dummy-entry-1".getBytes(Encoding));

        // read-position has not been moved
        assertFalse(c1.checkAndUpdateReadPositionChanged());

        List<Entry> entries = c1.readEntries(2);
        entries.forEach(e -> {
            try {
                c1.markDelete(e.getPosition());
                e.release();
            } catch (Exception e1) {
                // Ok
            }
        });

        // read-position is moved
        assertTrue(c1.checkAndUpdateReadPositionChanged());
        // read-position has not been moved since last read
        assertFalse(c1.checkAndUpdateReadPositionChanged());

        c1.close();
        ledger.close();

        ledger = factory.open("my_test_ledger", new ManagedLedgerConfig());
        // recover cursor
        ManagedCursor c2 = ledger.openCursor("c1");
        assertTrue(c2.checkAndUpdateReadPositionChanged());
        assertFalse(c2.checkAndUpdateReadPositionChanged());

        entries = c2.readEntries(2);
        entries.forEach(e -> {
            try {
                c2.markDelete(e.getPosition());
                e.release();
            } catch (Exception e1) {
                // Ok
            }
        });

        assertTrue(c2.checkAndUpdateReadPositionChanged());
        // returns true because read-position is on tail
        assertTrue(c2.checkAndUpdateReadPositionChanged());
        assertTrue(c2.checkAndUpdateReadPositionChanged());

        ledger.close();
    }


    @Test
    public void testCursorGetBacklog() throws Exception {
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        managedLedgerConfig.setMaxEntriesPerLedger(2);
        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("get-backlog", managedLedgerConfig);
        ManagedCursor managedCursor = ledger.openCursor("test");

        Position position = ledger.addEntry("test".getBytes(Encoding));
        ledger.addEntry("test".getBytes(Encoding));
        Position position1 = ledger.addEntry("test".getBytes(Encoding));
        ledger.addEntry("test".getBytes(Encoding));

        Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(true), 4);
        Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(false), 4);
        Field field = ManagedLedgerImpl.class.getDeclaredField("ledgers");
        field.setAccessible(true);

        ((ConcurrentSkipListMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo>) field.get(ledger)).remove(position.getLedgerId());
        field = ManagedCursorImpl.class.getDeclaredField("markDeletePosition");
        field.setAccessible(true);
        field.set(managedCursor, PositionImpl.get(position1.getLedgerId(), -1));


        Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(true), 2);
        Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(false), 4);
    }

    @Test
    public void testCursorNoRolloverIfNoMetadataSession() throws Exception {
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        managedLedgerConfig.setMaxEntriesPerLedger(2);
        managedLedgerConfig.setMetadataMaxEntriesPerLedger(2);
        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
        managedLedgerConfig.setThrottleMarkDelete(0);
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("testCursorNoRolloverIfNoMetadataSession", managedLedgerConfig);
        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test");

        List<Position> positions = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            positions.add(ledger.addEntry("test".getBytes(Encoding)));
        }

        cursor.delete(positions.get(0));

        long initialLedgerId = cursor.getCursorLedger();

        metadataStore.triggerSessionEvent(SessionEvent.SessionLost);

        for (int i = 1; i < 10; i++) {
            cursor.delete(positions.get(i));
        }

        assertEquals(cursor.getCursorLedger(), initialLedgerId);

        // After the session gets reestablished, the rollover should restart
        metadataStore.triggerSessionEvent(SessionEvent.SessionReestablished);

        for (int i = 0; i < 10; i++) {
            Position p = ledger.addEntry("test".getBytes(Encoding));
            cursor.delete(p);
        }

        assertNotEquals(cursor.getCursorLedger(), initialLedgerId);
    }

    @Test
    public void testReadEmptyEntryList() throws Exception {
        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
        managedLedgerConfig.setMaxEntriesPerLedger(1);
        managedLedgerConfig.setMetadataMaxEntriesPerLedger(1);
        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory
                .open("testReadEmptyEntryList", managedLedgerConfig);
        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test");

        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
        ledger.rollCurrentLedgerIfFull();

        AtomicBoolean flag = new AtomicBoolean();
        flag.set(false);
        ReadEntriesCallback callback = new ReadEntriesCallback() {
            @Override
            public void readEntriesComplete(List<Entry> entries, Object ctx) {
                if (entries.size() == 0) {
                    flag.set(true);
                }
            }

            @Override
            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {

            }
        };

        // op readPosition is bigger than maxReadPosition
        OpReadEntry opReadEntry = OpReadEntry.create(cursor, ledger.lastConfirmedEntry, 10, callback,
                null, PositionImpl.get(lastPosition.getLedgerId(), -1));
        Field field = ManagedCursorImpl.class.getDeclaredField("readPosition");
        field.setAccessible(true);
        field.set(cursor, PositionImpl.earliest);
        ledger.asyncReadEntries(opReadEntry);

        // when readPosition is bigger than maxReadPosition, should complete the opReadEntry
        Awaitility.await().untilAsserted(() -> assertTrue(flag.get()));
    }

    private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
}