[fix] Fix cursor deadlock when skipping non-recoverable entries during mark-delete persistence

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

@@ -3185,7 +3185,7 @@ public void deleteFailed(ManagedLedgerException ex, Object ctx) {
      * Manually acknowledge all entries from startPosition to endPosition.
      * - Since this is an uncommon event, we focus on maintainability. So we do not modify
      *   {@link #individualDeletedMessages} and {@link #batchDeletedIndexes}, but call
-     *   {@link #asyncDelete(Position, AsyncCallbacks.DeleteCallback, Object)}.
+     *   {@link #asyncDelete(Iterable, AsyncCallbacks.DeleteCallback, Object)}.
      * - This method is valid regardless of the consumer ACK type.
      * - If there is a consumer ack request after this event, it will also work.
      */
@@ -3202,33 +3202,26 @@ public void skipNonRecoverableEntries(Position startPosition, Position endPositi
             return;
         }
 
-        lock.writeLock().lock();
         log.warn()
                 .attr("ledgerId", ledgerId)
                 .attr("startEntryId", startEntryId)
                 .attr("endEntryId", endEntryId)
                 .log("Entries are lost, auto-acknowledging in subscription (autoSkipNonRecoverableData=true)");
-        try {
-            for (long i = startEntryId; i < endEntryId; i++) {
-                if (!individualDeletedMessages.contains(ledgerId, i)) {
-                    asyncDelete(PositionFactory.create(ledgerId, i), new AsyncCallbacks.DeleteCallback() {
-                        @Override
-                        public void deleteComplete(Object ctx) {
-                            // ignore.
-                        }
+        asyncDelete(() -> LongStream.range(startEntryId, endEntryId)
+                        .mapToObj(i -> PositionFactory.create(ledgerId, i)).iterator(),
+                new AsyncCallbacks.DeleteCallback() {
+                    @Override
+                    public void deleteComplete(Object ctx) {
+                        // ignore.
+                    }
 
-                        @Override
-                        public void deleteFailed(ManagedLedgerException ex, Object ctx) {
-                            // The method internalMarkDelete already handled the failure operation. We only need to
-                            // make sure the memory state is updated.
-                            // If the broker crashed, the non-recoverable ledger will be detected again.
-                        }
-                    }, null);
-                }
-            }
-        } finally {
-            lock.writeLock().unlock();
-        }
+                    @Override
+                    public void deleteFailed(ManagedLedgerException ex, Object ctx) {
+                        // The method internalMarkDelete already handled the failure operation. We only need to
+                        // make sure the memory state is updated.
+                        // If the broker crashed, the non-recoverable ledger will be detected again.
+                    }
+                }, null);
     }
 
     // //////////////////////////////////////////////////

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

@@ -6240,6 +6240,44 @@ public void testSkipNonRecoverableEntries() throws ManagedLedgerException, Inter
         ledger.close();
     }
 
+    @Test(timeOut = 20000)
+    public void testSkipNonRecoverableEntriesDoesNotHoldCursorLockWhilePersisting() throws Exception {
+        ManagedLedger ledger = factory.open("testSkipNonRecoverableEntriesDoesNotHoldCursorLockWhilePersisting",
+                new ManagedLedgerConfig());
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("my-cursor");
+        Position firstPosition = ledger.addEntry("entry-0".getBytes(Encoding));
+        Position nextPosition = ledger.addEntry("entry-1".getBytes(Encoding));
+
+        AtomicBoolean readLockAvailable = new AtomicBoolean();
+        Thread skipThread = new Thread(
+                () -> cursor.skipNonRecoverableEntries(firstPosition, nextPosition),
+                "skip-non-recoverable-entries");
+        synchronized (cursor.pendingMarkDeleteOps) {
+            skipThread.start();
+            Awaitility.await().atMost(Duration.ofSeconds(5))
+                    .until(() -> skipThread.getState() == Thread.State.BLOCKED
+                            && Arrays.stream(skipThread.getStackTrace())
+                                    .anyMatch(frame -> frame.getMethodName().equals("internalAsyncMarkDelete")));
+            if (cursor.lock.readLock().tryLock()) {
+                try {
+                    readLockAvailable.set(true);
+                } finally {
+                    cursor.lock.readLock().unlock();
+                }
+            }
+        }
+
+        skipThread.join(TimeUnit.SECONDS.toMillis(5));
+        assertFalse(skipThread.isAlive(), "skip thread did not finish");
+        Awaitility.await().atMost(Duration.ofSeconds(5))
+                .until(() -> cursor.getStats().getPersistLedgerSucceed() > 0);
+        assertTrue(readLockAvailable.get(),
+                "skipNonRecoverableEntries must not hold the cursor lock while waiting to persist mark-delete");
+
+        cursor.close();
+        ledger.close();
+    }
+
     @Test
     public void testRecoverCursorWithTerminateManagedLedger() throws Exception {
         String mlName = "my_test_ledger";