Optimize filtering deleted messages

lhotari · Nov 13, 2024 · a0d0b54 · a0d0b54
1 parent a126e40
commit a0d0b54
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 10 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
@@ -19,6 +19,8 @@
 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;
@@ -893,6 +895,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);

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -35,6 +35,7 @@
 import java.time.Clock;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -1565,14 +1566,7 @@ public Set<? extends Position> asyncReplayEntries(Set<? extends Position> positi
             return Collections.emptySet();
         }
 
-        // filters out messages which are already acknowledged
-        Set<Position> alreadyAcknowledgedPositions = new HashSet<>();
-        lock.readLock().lock();
-        try {
-            positions.stream().filter(this::isMessageDeleted).forEach(alreadyAcknowledgedPositions::add);
-        } finally {
-            lock.readLock().unlock();
-        }
+        Set<Position> alreadyAcknowledgedPositions = filterDeletedMessages(positions);
 
         final int totalValidPositions = positions.size() - alreadyAcknowledgedPositions.size();
         if (totalValidPositions == 0) {
@@ -1634,6 +1628,25 @@ public String toString() {
         return alreadyAcknowledgedPositions;
     }
 
+    @Override
+    public Set<Position> filterDeletedMessages(Collection<? extends Position> positions) {
+        Set<Position> deletedMessages = new HashSet<>();
+        // acquire a read lock once for all the positions
+        lock.readLock().lock();
+        try {
+            // prefer for loop to avoid creating stream related instances
+            for (Position position : positions) {
+                // call the internal method to avoid acquiring read lock multiple times
+                if (internalIsMessageDeleted(position)) {
+                    deletedMessages.add(position);
+                }
+            }
+        } finally {
+            lock.readLock().unlock();
+        }
+        return deletedMessages;
+    }
+
     protected long getNumberOfEntries(Range<Position> range) {
         long allEntries = ledger.getNumberOfEntries(range);
 
@@ -3547,13 +3560,18 @@ public Position processIndividuallyDeletedMessagesAndGetMarkDeletedPosition(
     public boolean isMessageDeleted(Position position) {
         lock.readLock().lock();
         try {
-            return position.compareTo(markDeletePosition) <= 0
-                    || individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId());
+            return internalIsMessageDeleted(position);
         } finally {
             lock.readLock().unlock();
         }
     }
 
+    // calling this method expects that the lock is already acquired
+    private boolean internalIsMessageDeleted(Position position) {
+        return position.compareTo(markDeletePosition) <= 0
+                || individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId());
+    }
+
     //this method will return a copy of the position's ack set
     @Override
     public long[] getBatchPositionAckSet(Position position) {