[fix][test] Modify code for debug reason

Author: oneby-wang

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

@@ -1243,14 +1243,15 @@ public boolean hasMoreEntries() {
 
     @Override
     public long getNumberOfEntries() {
-        if (readPosition.compareTo(ledger.getLastPosition().getNext()) > 0) {
+        Position lastPosition = ledger.getLastPosition();
+        if (readPosition.compareTo(lastPosition.getNext()) > 0) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] [{}] Read position {} is ahead of last position {}. There are no entries to read",
-                        ledger.getName(), name, readPosition, ledger.getLastPosition());
+                        ledger.getName(), name, readPosition, lastPosition);
             }
             return 0;
         } else {
-            return getNumberOfEntries(Range.closedOpen(readPosition, ledger.getLastPosition().getNext()));
+            return getNumberOfEntries(Range.closedOpen(readPosition, lastPosition.getNext()));
         }
     }
 
@@ -2182,25 +2183,41 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
         }
 
         Position newPosition = ackBatchPosition(position);
-        if (ledger.getLastConfirmedEntry().compareTo(newPosition) < 0) {
+        Position markDeletePos = markDeletePosition;
+        Position lastConfirmedEntry = ledger.getLastConfirmedEntry();
+        if (lastConfirmedEntry.compareTo(newPosition) < 0) {
+            // Extract local variables to here for debug reason.
             boolean shouldCursorMoveForward = false;
+            long ledgerEntries = 0;
+            Long nextValidLedger = null;
             try {
-                long ledgerEntries = ledger.getLedgerInfo(markDeletePosition.getLedgerId()).get().getEntries();
-                Long nextValidLedger = ledger.getNextValidLedger(ledger.getLastConfirmedEntry().getLedgerId());
-                shouldCursorMoveForward = nextValidLedger != null
-                        && (markDeletePosition.getEntryId() + 1 >= ledgerEntries)
-                        && (newPosition.getLedgerId() == nextValidLedger);
+                LedgerInfo markDeleteLedgerInfo = ledger.getLedgerInfo(markDeletePos.getLedgerId()).get();
+                LedgerInfo newPositionLedgerInfo = null;
+                if (markDeleteLedgerInfo == null) {
+                    newPositionLedgerInfo = ledger.getLedgerInfo(newPosition.getLedgerId()).get();
+                    shouldCursorMoveForward = newPositionLedgerInfo != null
+                            && newPosition.getEntryId() + 1 == newPositionLedgerInfo.getEntries();
+                } else {
+                    ledgerEntries = markDeleteLedgerInfo.getEntries();
+                    nextValidLedger = ledger.getNextValidLedger(lastConfirmedEntry.getLedgerId());
+                    shouldCursorMoveForward = nextValidLedger != null
+                            && (markDeletePos.getEntryId() + 1 >= ledgerEntries)
+                            && (newPosition.getLedgerId() == nextValidLedger);
+                }
+                if (!shouldCursorMoveForward) {
+                    log.info("test");
+                }
             } catch (Exception e) {
                 log.warn("Failed to get ledger entries while setting mark-delete-position", e);
             }
 
             if (shouldCursorMoveForward) {
                 log.info("[{}] move mark-delete-position from {} to {} since all the entries have been consumed",
-                        ledger.getName(), markDeletePosition, newPosition);
+                        ledger.getName(), markDeletePos, newPosition);
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] Failed mark delete due to invalid markDelete {} is ahead of last-confirmed-entry {}"
-                             + " for cursor [{}]", ledger.getName(), position, ledger.getLastConfirmedEntry(), name);
+                             + " for cursor [{}]", ledger.getName(), position, lastConfirmedEntry, name);
                 }
                 callback.markDeleteFailed(new ManagedLedgerException("Invalid mark deleted position"), ctx);
                 return;

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

@@ -2717,8 +2717,8 @@ public CompletableFuture<Void> maybeUpdateCursorBeforeTrimmingConsumedLedger() {
 
             // Snapshot cursor.getMarkDeletedPosition() into a local variable to avoid race condition.
             Position markDeletedPosition = cursor.getMarkDeletedPosition();
-            Position lastAckedPosition = cursor.getPersistentMarkDeletedPosition() != null
-                    ? cursor.getPersistentMarkDeletedPosition() : markDeletedPosition;
+            // When doing mark delete operation, persistentMarkDeletedPosition is updated after markDeletedPosition
+            Position lastAckedPosition = markDeletedPosition;
             LedgerInfo curPointedLedger   = ledgers.get(lastAckedPosition.getLedgerId());
             LedgerInfo nextPointedLedger = Optional.ofNullable(ledgers.higherEntry(lastAckedPosition.getLedgerId()))
                     .map(Map.Entry::getValue).orElse(null);
@@ -2767,14 +2767,14 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
                         cursor, markDeletedPosition);
                 future.complete(null);
             } else {
-                // Should not happen
+                // Could not happen
                 log.warn("Trying to mark delete an already mark-deleted position. Current mark-delete:"
                         + " {} -- attempted position: {}", markDeletedPosition, 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));
+                // future.complete(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

@@ -3787,7 +3787,7 @@ public void testExpiredLedgerDeletionAfterManagedLedgerRestart() throws Exceptio
     }
 
     // Huh, maybeUpdateCursorBeforeTrimmingConsumedLedger seems not working well.
-    @Test(timeOut = 20000)
+    @Test(timeOut = 20000, invocationCount = 200)
     public void testNeverThrowsMarkDeletingMarkedPositionInMaybeUpdateCursorBeforeTrimmingConsumedLedger()
             throws Exception {
         ManagedLedgerConfig config = new ManagedLedgerConfig();
@@ -3822,6 +3822,7 @@ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
                     cursor.asyncMarkDelete(position, new MarkDeleteCallback() {
                         @Override
                         public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                            fail("Should not happen");
                         }
 
                         @Override