From 2d2f1591a13ecd5cbf9431257ee11b2b3ccc722a Mon Sep 17 00:00:00 2001
From: Denovo1998 <liusinan1998@gmail.com>
Date: Mon, 18 Nov 2024 21:12:23 +0800
Subject: [PATCH] avoid nested read-write locks

---
 .../bookkeeper/mledger/impl/ManagedCursorImpl.java | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

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
index 7c0d13108b1c4..478c6a1b37976 100644
--- 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
@@ -1569,7 +1569,7 @@ public Set<? extends Position> asyncReplayEntries(Set<? extends Position> positi
         Set<Position> alreadyAcknowledgedPositions = new HashSet<>();
         lock.readLock().lock();
         try {
-            positions.stream().filter(this::isMessageDeleted).forEach(alreadyAcknowledgedPositions::add);
+            positions.stream().filter(this::internalIsMessageDeleted).forEach(alreadyAcknowledgedPositions::add);
         } finally {
             lock.readLock().unlock();
         }
@@ -2345,7 +2345,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
                     return;
                 }
 
-                if (isMessageDeleted(position)) {
+                if (internalIsMessageDeleted(position)) {
                     if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
                         BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
                         if (bitSetRecyclable != null) {
@@ -3543,13 +3543,19 @@ 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();
         }
     }
 
+    // When this method is called while the external has already acquired a write lock or a read lock,
+    // it avoids unnecessary lock nesting.
+    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) {