Future completed twice in the method of impl.MLPendingAckStore#closeA…

…sync (#12362) (cherry picked from commit 10371ee)
apache · Dec 20, 2021 · 86681ae · 86681ae
1 parent bc50bac
commit 86681ae
Showing 1 changed file with 17 additions and 6 deletions.
diff --git a/...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java b/...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
@@ -120,12 +120,23 @@ public CompletableFuture<Void> closeAsync() {
         cursor.asyncClose(new AsyncCallbacks.CloseCallback() {
             @Override
             public void closeComplete(Object ctx) {
-                try {
-                    managedLedger.close();
-                } catch (Exception e) {
-                    completableFuture.completeExceptionally(e);
-                }
-                completableFuture.complete(null);
+                managedLedger.asyncClose(new AsyncCallbacks.CloseCallback() {
+
+                    @Override
+                    public void closeComplete(Object ctx) {
+                        if (log.isDebugEnabled()) {
+                            log.debug("[{}][{}] MLPendingAckStore closed successfully！", managedLedger.getName(), ctx);
+                        }
+                        completableFuture.complete(null);
+                    }
+
+                    @Override
+                    public void closeFailed(ManagedLedgerException exception, Object ctx) {
+                        log.error("[{}][{}] MLPendingAckStore closed failed,exception={}", managedLedger.getName(),
+                                ctx, exception);
+                        completableFuture.completeExceptionally(exception);
+                    }
+                }, ctx);
             }
 
             @Override