[fix][ml]Received more than once callback when calling cursor.delete (apache#24405)

(cherry picked from commit 72953bf)

Author: poorbarcode

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

@@ -2349,7 +2349,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
         PositionImpl newMarkDeletePosition = null;
 
         lock.writeLock().lock();
-
+        boolean skipMarkDeleteBecauseAckedNothing = false;
         try {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] [{}] Deleting individual messages at {}. Current status: {} - md-position: {}",
@@ -2418,6 +2418,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
 
             if (individualDeletedMessages.isEmpty()) {
                 // No changes to individually deleted messages, so nothing to do at this point
+                skipMarkDeleteBecauseAckedNothing = true;
                 return;
             }
 
@@ -2435,6 +2436,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
 
             if (range == null) {
                 // The set was completely cleaned up now
+                skipMarkDeleteBecauseAckedNothing = true;
                 return;
             }
 
@@ -2461,9 +2463,8 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
             callback.deleteFailed(getManagedLedgerException(e), ctx);
             return;
         } finally {
-            boolean empty = individualDeletedMessages.isEmpty();
             lock.writeLock().unlock();
-            if (empty) {
+            if (skipMarkDeleteBecauseAckedNothing) {
                 callback.deleteComplete(ctx);
             }
         }

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

@@ -5233,5 +5233,33 @@ public void testDeleteBatchedMessageWithEmptyAckSet() throws Exception {
         ml.delete();
     }
 
+    @Test
+    public void testCallbackTimes() throws Exception {
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open("testCallbackTimes");
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ml.openCursor("c1");
+        Position position1 = ml.addEntry(new byte[1]);
+        Position position2 = ml.addEntry(new byte[2]);
+        AtomicInteger executedCallbackTimes = new AtomicInteger();
+        cursor.asyncDelete(Arrays.asList(position1, position2), new DeleteCallback() {
+            @Override
+            public void deleteComplete(Object ctx) {
+                executedCallbackTimes.incrementAndGet();
+            }
+
+            @Override
+            public void deleteFailed(ManagedLedgerException exception, Object ctx) {
+                executedCallbackTimes.incrementAndGet();
+            }
+        }, new Object());
+        // Verify that the executed count of callback is "1".
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(executedCallbackTimes.get() > 0);
+        });
+        Thread.sleep(2000);
+        assertEquals(executedCallbackTimes.get(), 1);
+        // cleanup.
+        ml.delete();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
 }