lh fix individual read deduplication and limiting

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/CachedEntry.java

@@ -0,0 +1,26 @@
+/*
+ * 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;
+
+import io.netty.util.ReferenceCounted;
+
+public interface CachedEntry extends Entry, ReferenceCounted {
+    boolean matchesKey(Position key);
+    boolean canEvict();
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java

@@ -32,8 +32,10 @@ public interface Entry {
     /**
      * @return the data
      */
+    @Deprecated
     byte[] getData();
 
+    @Deprecated
     byte[] getDataAndRelease();
 
     /**
@@ -66,4 +68,8 @@ public interface Entry {
      * of data reached to 0).
      */
     boolean release();
+
+    default EntryReadCountHandler getReadCountHandler() {
+        return null;
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/EntryReadCountHandler.java

@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+public interface EntryReadCountHandler {
+    int getExpectedReadCount();
+    boolean incrementExpectedReadCount();
+    void markRead();
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

@@ -19,10 +19,13 @@
 package org.apache.bookkeeper.mledger;
 
 import com.google.common.collect.Range;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.SortedSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
@@ -718,6 +721,7 @@ default void asyncFindNewestMatching(FindPositionConstraint constraint, Predicat
      * @throws InterruptedException
      * @throws ManagedLedgerException
      */
+    @Deprecated
     List<Entry> replayEntries(Set<? extends Position> positions)
             throws InterruptedException, ManagedLedgerException;
 
@@ -733,8 +737,10 @@ List<Entry> replayEntries(Set<? extends Position> positions)
      * @return skipped positions
      *              set of positions which are already deleted/acknowledged and skipped while replaying them
      */
-    Set<? extends Position> asyncReplayEntries(
-        Set<? extends Position> positions, ReadEntriesCallback callback, Object ctx);
+    default Set<? extends Position> asyncReplayEntries(final Set<? extends Position> positions,
+                                                      ReadEntriesCallback callback, Object ctx) {
+        return asyncReplayEntries(positions, callback, ctx, false);
+    }
 
     /**
      * Read the specified set of positions from ManagedLedger.
@@ -753,6 +759,28 @@ Set<? extends Position> asyncReplayEntries(
     Set<? extends Position> asyncReplayEntries(
             Set<? extends Position> positions, ReadEntriesCallback callback, Object ctx, boolean sortEntries);
 
+    /**
+     * Read the specified set of positions from ManagedLedger in ranges.
+     * This method is used to read entries in ranges to avoid reading all entries at once.
+     *
+     * @param positions
+     *            set of positions to read
+     * @param callback
+     *            callback object returning the result of each range
+     * @param ctx
+     *            opaque context
+     * @param invokeCallbacksInOrder
+     *            when true, the callback will be invoked in order of the positions, otherwise the callback will be
+     *            invoked in the order of the completion of the range read.
+     * @return skipped positions
+     *              set of positions which are already deleted/acknowledged and skipped while replaying them
+     */
+    default Set<? extends Position> asyncReplayEntriesInRanges(SortedSet<? extends Position> positions,
+                                                               ManagedCursorReplayReadEntriesCallback callback,
+                                                               Object ctx, boolean invokeCallbacksInOrder) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
     /**
      * Close the cursor and releases the associated resources.
      *
@@ -918,6 +946,24 @@ default ManagedCursorAttributes getManagedCursorAttributes() {
 
     boolean isMessageDeleted(Position position);
 
+    /**
+     * Returns the deleted messages from the given positions.
+     * Implementation classes can override this method to provide a more efficient way to filter deleted messages.
+     *
+     * @param positions the positions to filter
+     * @return the set of deleted positions
+     */
+    default Set<Position> filterDeletedMessages(Collection<? extends Position> positions) {
+        Set<Position> deletedPositions = new HashSet<>();
+        // prefer for loop to avoid creating stream related instances
+        for (Position position : positions) {
+            if (isMessageDeleted(position)) {
+                deletedPositions.add(position);
+            }
+        }
+        return deletedPositions;
+    }
+
     ManagedCursor duplicateNonDurableCursor(String nonDurableCursorName) throws ManagedLedgerException;
 
     long[] getBatchPositionAckSet(Position position);

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorReplayReadEntriesCallback.java

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+import java.util.List;
+
+public interface ManagedCursorReplayReadEntriesCallback {
+    void readEntriesComplete(ManagedCursorReplayReadRange range, boolean isLast, List<Entry> entries, Object ctx);
+
+    void readEntriesFailed(ManagedCursorReplayReadRange range, boolean isLast, ManagedLedgerException exception,
+                           Object ctx);
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorReplayReadRange.java

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+public interface ManagedCursorReplayReadRange extends Comparable<ManagedCursorReplayReadRange> {
+    int rangeIndex();
+
+    int totalRanges();
+
+    Position startPosition();
+
+    Position lastPosition();
+
+    default int size() {
+        if (startPosition().getLedgerId() != lastPosition().getLedgerId()) {
+            throw new IllegalStateException("Cannot calculate size for range spanning multiple ledgers");
+        }
+        return (int) (lastPosition().getEntryId() - startPosition().getEntryId() + 1);
+    }
+
+    @Override
+    default int compareTo(ManagedCursorReplayReadRange o) {
+        return Integer.compare(rangeIndex(), o.rangeIndex());
+    }
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -83,6 +83,9 @@ public class ManagedLedgerConfig {
     @Getter
     @Setter
     private boolean cacheEvictionByMarkDeletedPosition = false;
+    @Getter
+    @Setter
+    private boolean cacheEvictionByExpectedReadCount = true;
     private int minimumBacklogCursorsForCaching = 0;
     private int minimumBacklogEntriesForCaching = 1000;
     private int maxBacklogBetweenCursorsForCaching = 1000;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java

@@ -61,6 +61,18 @@ public class ManagedLedgerFactoryConfig {
      */
     private long managedLedgerMaxReadsInFlightSize = 0;
 
+    /**
+     * Maximum time to wait for acquiring permits for max reads in flight when managedLedgerMaxReadsInFlightSizeInMB is
+     * set (>0) and the limit is reached.
+     */
+    private long managedLedgerMaxReadsInFlightPermitsAcquireTimeoutMillis = 60000;
+
+    /**
+     * Maximum number of reads that can be queued for acquiring permits for max reads in flight when
+     * managedLedgerMaxReadsInFlightSizeInMB is set (>0) and the limit is reached.
+     */
+    private int managedLedgerMaxReadsInFlightPermitsAcquireQueueSize = 10000;
+
     /**
      * Whether trace managed ledger task execution time.
      */