[fix][test] Rollback some codes for test

Author: oneby-wang

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

@@ -483,15 +483,20 @@ public void initializeComplete() {
                                 pendingInitializeLedgers.remove(name, pendingLedger);
                                 future.complete(newledger);
 
-                                // May need to update the cursor position and wait them finished
-                                newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger().whenComplete((__, ex) -> {
-                                    // ignore ex since it is handled in maybeUpdateCursorBeforeTrimmingConsumedLedger
-                                    future.complete(newledger);
-                                    // May need to trigger offloading
-                                    if (config.isTriggerOffloadOnTopicLoad()) {
-                                        newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE);
-                                    }
-                                });
+                                newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger();
+                                if (config.isTriggerOffloadOnTopicLoad()) {
+                                    newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE);
+                                }
+
+//                                // May need to update the cursor position and wait them finished
+//                                newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger().whenComplete((__, ex) -> {
+//                                    // ignore ex since it is handled in maybeUpdateCursorBeforeTrimmingConsumedLedger
+//                                    future.complete(newledger);
+//                                    // May need to trigger offloading
+//                                    if (config.isTriggerOffloadOnTopicLoad()) {
+//                                        newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE);
+//                                    }
+//                                });
                             }
 
                             @Override

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

@@ -2716,7 +2716,7 @@ public CompletableFuture<Void> maybeUpdateCursorBeforeTrimmingConsumedLedger() {
             cursorMarkDeleteFutures.add(future);
 
             // Snapshot cursor.getMarkDeletedPosition() into a local variable to avoid race condition.
-            Position markDeletedPosition = PositionFactory.create(cursor.getMarkDeletedPosition());
+            Position markDeletedPosition = cursor.getMarkDeletedPosition();
             Position lastAckedPosition = cursor.getPersistentMarkDeletedPosition() != null
                     ? cursor.getPersistentMarkDeletedPosition() : markDeletedPosition;
             LedgerInfo curPointedLedger   = ledgers.get(lastAckedPosition.getLedgerId());
@@ -2770,10 +2770,11 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                 // Should not happen
                 log.warn("Trying to mark delete an already mark-deleted position. Current mark-delete:"
                         + " {} -- attempted position: {}", markDeletedPosition, lastAckedPosition);
-                future.completeExceptionally(new ManagedLedgerException(
-                        "Trying to mark delete an already mark-deleted position when update cursor. Current "
-                                + "mark-delete: " + markDeletedPosition + " -- attempted mark delete: "
-                                + lastAckedPosition));
+                future.completeExceptionally(null);
+//                future.completeExceptionally(new ManagedLedgerException(
+//                        "Trying to mark delete an already mark-deleted position when update cursor. Current "
+//                                + "mark-delete: " + markDeletedPosition + " -- attempted mark delete: "
+//                                + lastAckedPosition));
             }
         }
         return FutureUtil.waitForAll(cursorMarkDeleteFutures);

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

@@ -56,6 +56,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -69,6 +70,7 @@
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ExecutorService;
@@ -3784,35 +3786,39 @@ public void testExpiredLedgerDeletionAfterManagedLedgerRestart() throws Exceptio
         });
     }
 
+    // Huh, maybeUpdateCursorBeforeTrimmingConsumedLedger seems not working well.
     @Test(timeOut = 20000)
     public void testNeverThrowsMarkDeletingMarkedPositionInMaybeUpdateCursorBeforeTrimmingConsumedLedger()
             throws Exception {
         ManagedLedgerConfig config = new ManagedLedgerConfig();
         initManagedLedgerConfig(config);
         config.setMaxEntriesPerLedger(1);
         int entryNum = 100;
-        config.setMaxEntriesPerLedger(entryNum);
 
-        ManagedLedgerImpl managedLedger =
+        ManagedLedgerImpl realManagedLedger =
                 (ManagedLedgerImpl) factory.open("maybeUpdateCursorBeforeTrimmingConsumed_ledger", config);
+        ManagedLedgerImpl managedLedger = spy(realManagedLedger);
         ManagedCursor cursor = managedLedger.openCursor("c1");
 
+        Deque<CompletableFuture<Void>> futures = new ConcurrentLinkedDeque<>();
+        doAnswer(invocation -> {
+            CompletableFuture<Void> result = (CompletableFuture<Void>) invocation.callRealMethod();
+            futures.offer(result);
+            return result;
+        }).when(managedLedger).maybeUpdateCursorBeforeTrimmingConsumedLedger();
+
         final CountDownLatch latch = new CountDownLatch(entryNum);
-        List<CompletableFuture<Void>> updateCursorFutures = new ArrayList<>(entryNum);
-        // Three asyncMarkDelete operations running concurrently:
+        // Two asyncMarkDelete operations running concurrently:
         //   1. ledger rollover triggered maybeUpdateCursorBeforeTrimmingConsumedLedger.
         //   2. user triggered asyncMarkDelete.
-        //   3. user triggered maybeUpdateCursorBeforeTrimmingConsumedLedger.
         for (int i = 0; i < entryNum; i++) {
-            CountDownLatch taskFireLatch = new CountDownLatch(1);
             managedLedger.asyncAddEntry("entry".getBytes(Encoding), new AddEntryCallback() {
                 @Override
                 public void addFailed(ManagedLedgerException exception, Object ctx) {
                 }
 
                 @Override
                 public void addComplete(Position position, ByteBuf entryData, Object ctx) {
-                    taskFireLatch.countDown();
                     cursor.asyncMarkDelete(position, new MarkDeleteCallback() {
                         @Override
                         public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
@@ -3826,15 +3832,13 @@ public void markDeleteComplete(Object ctx) {
 
                 }
             }, null);
-            taskFireLatch.await();
-            CompletableFuture<Void> future = managedLedger.maybeUpdateCursorBeforeTrimmingConsumedLedger();
-            updateCursorFutures.add(future);
         }
 
         latch.await();
         assertEquals(cursor.getNumberOfEntries(), 0);
+
         // Will not throw exception
-        FutureUtil.waitForAll(updateCursorFutures).get();
+        FutureUtil.waitForAll(futures).get();
     }
 
     @Test(timeOut = 20000)