[fix][broker] Fix deadlock in PendingAckHandleImpl (#18989)

apache · Dec 20, 2022 · 22866bd · 22866bd
1 parent 37ea179
commit 22866bd
Showing 1 changed file with 10 additions and 12 deletions.
diff --git a/.../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java b/.../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
@@ -155,20 +155,20 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
                         .getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
 
         pendingAckStoreProvider.checkInitializedBefore(persistentSubscription)
-                .thenAccept(init -> {
+                .thenAcceptAsync(init -> {
                     if (init) {
                         initPendingAckStore();
                     } else {
                         completeHandleFuture();
                     }
-                })
-                .exceptionally(e -> {
+                }, internalPinnedExecutor)
+                .exceptionallyAsync(e -> {
                     Throwable t = FutureUtil.unwrapCompletionException(e);
                     changeToErrorState();
                     exceptionHandleFuture(t);
                     this.pendingAckStoreFuture.completeExceptionally(t);
                     return null;
-                });
+                }, internalPinnedExecutor);
     }
 
     private void initPendingAckStore() {
@@ -937,18 +937,16 @@ public TransactionPendingAckStats getStats(boolean lowWaterMarks) {
         return transactionPendingAckStats;
     }
 
-    public synchronized void completeHandleFuture() {
-        if (!this.pendingAckHandleCompletableFuture.isDone()) {
-            this.pendingAckHandleCompletableFuture.complete(PendingAckHandleImpl.this);
-        }
-        if (recoverTime.getRecoverStartTime() != 0L) {
+    public void completeHandleFuture() {
+        this.pendingAckHandleCompletableFuture.complete(PendingAckHandleImpl.this);
+        if (recoverTime.getRecoverStartTime() != 0L && recoverTime.getRecoverEndTime() == 0L) {
             recoverTime.setRecoverEndTime(System.currentTimeMillis());
         }
     }
 
-    public synchronized void exceptionHandleFuture(Throwable t) {
-        if (!this.pendingAckHandleCompletableFuture.isDone()) {
-            this.pendingAckHandleCompletableFuture.completeExceptionally(t);
+    public void exceptionHandleFuture(Throwable t) {
+        final boolean completedNow = this.pendingAckHandleCompletableFuture.completeExceptionally(t);
+        if (completedNow) {
             recoverTime.setRecoverEndTime(System.currentTimeMillis());
         }
     }