fix testcase

Author: wuzhanpeng

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

@@ -3183,10 +3183,16 @@ private void cleanupOffloaded(long ledgerId, UUID uuid, String offloadDriverName
         metadataMap.putAll(offloadDriverMetadata);
         metadataMap.put("ManagedLedgerName", name);
 
-        Retries.run(Backoff.exponentialJittered(TimeUnit.SECONDS.toMillis(1), TimeUnit.SECONDS.toHours(1)).limit(10),
-                Retries.NonFatalPredicate,
-                () -> config.getLedgerOffloader().deleteOffloaded(ledgerId, uuid, metadataMap),
-                scheduledExecutor, name).whenComplete((ignored, exception) -> {
+        Retries
+                // The purpose of not specifying the scheduler's key explicitly here is to avoid deadlock.
+                // Otherwise when the caller of this method and the retry task are both running in the same
+                // ordering thread, there will be a risk of race-condition.
+                .run(
+                        Backoff.exponentialJittered(TimeUnit.SECONDS.toMillis(1), TimeUnit.SECONDS.toHours(1)).limit(10),
+                        Retries.NonFatalPredicate,
+                        () -> config.getLedgerOffloader().deleteOffloaded(ledgerId, uuid, metadataMap),
+                        scheduledExecutor)
+                .whenComplete((ignored, exception) -> {
                     if (exception != null) {
                         log.warn("[{}] Error cleaning up offload for {}, (cleanup reason: {})",
                                 name, ledgerId, cleanupReason, exception);
@@ -4134,6 +4140,7 @@ public void removeAllDeletableLedgers() {
         final CountDownLatch counter = new CountDownLatch(deletableLedgers.size() + deletableOffloadedLedgers.size());
 
         Set<Long> finishedDeletedLedgers = ConcurrentHashMap.newKeySet();
+        Set<Long> finishedDeletedOffloadedLedgers = ConcurrentHashMap.newKeySet();
         Set<Long> timeoutDeletedLedgers = ConcurrentHashMap.newKeySet();
 
         Set<Long> succeedDeletedLedgers = ConcurrentHashMap.newKeySet();
@@ -4170,7 +4177,7 @@ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) {
                         public void deleteLedgerComplete(Object ctx) {
                             ledgers.remove(deletableOffloadedLedger);
                             counter.countDown();
-                            finishedDeletedLedgers.add(deletableOffloadedLedger);
+                            finishedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
                             succeedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
                         }
 
@@ -4179,19 +4186,25 @@ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) {
                             log.warn("[{}] Failed to delete offloaded ledger:{} due to",
                                     name, deletableOffloadedLedger, exception);
                             counter.countDown();
-                            finishedDeletedLedgers.add(deletableOffloadedLedger);
+                            finishedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
                             failDeletedOffloadedLedgers.add(deletableOffloadedLedger);
                         }
                     });
         }
 
         try {
             if (!counter.await(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS)) {
-                for (Long ledgerId : Stream.concat(deletableLedgers.stream(), deletableOffloadedLedgers.stream())
-                        .collect(Collectors.toSet())) {
-                    if (!finishedDeletedLedgers.contains(ledgerId)) {
-                        log.warn("[{}] Failed to delete ledger:{} due to operation timeout", name, ledgerId);
-                        timeoutDeletedLedgers.add(ledgerId);
+                for (Long deletableLedger : deletableLedgers) {
+                    if (!finishedDeletedLedgers.contains(deletableLedger)) {
+                        log.warn("[{}] Failed to delete ledger:{} due to operation timeout", name, deletableLedger);
+                        timeoutDeletedLedgers.add(deletableLedger);
+                    }
+                }
+                for (Long deletableOffloadedLedger : deletableOffloadedLedgers) {
+                    if (!finishedDeletedOffloadedLedgers.contains(deletableOffloadedLedger)) {
+                        log.warn("[{}] Failed to delete offloaded ledger:{} due to operation timeout",
+                                name, deletableOffloadedLedger);
+                        timeoutDeletedLedgers.add(deletableOffloadedLedger);
                     }
                 }
             }