[pulsar] branch branch-2.10 updated: [fix][txn] Fix PendingAckHandleImpl when `pendingAckStoreProvider.checkInitializedBefore` failed (#18859)

[xiangying]

This is an automated email from the ASF dual-hosted git repository.

xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.10 by this push:
     new e5dedfe9953 [fix][txn] Fix PendingAckHandleImpl when 
`pendingAckStoreProvider.checkInitializedBefore` failed (#18859)
e5dedfe9953 is described below

commit e5dedfe99538a0dcdb12a7b8483259399ec16ed3
Author: Tao Jiuming <95597048+tjium...@users.noreply.github.com>
AuthorDate: Mon Dec 12 16:54:28 2022 +0800

    [fix][txn] Fix PendingAckHandleImpl when 
`pendingAckStoreProvider.checkInitializedBefore` failed (#18859)
    
    (cherry picked from commit 1be5a69a079594d8d96d2a0ab7ab8b389da8865e)
---
 .../pendingack/impl/PendingAckHandleImpl.java       | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
index 90bddef52d5..3422401ab83 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
@@ -141,13 +141,20 @@ public class PendingAckHandleImpl extends 
PendingAckHandleState implements Pendi
 
         this.pendingAckStoreProvider = this.persistentSubscription.getTopic()
                         
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
-        
pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init
 -> {
-            if (init) {
-                initPendingAckStore();
-            } else {
-                completeHandleFuture();
-            }
-        });
+
+        pendingAckStoreProvider.checkInitializedBefore(persistentSubscription)
+                .thenAccept(init -> {
+                    if (init) {
+                        initPendingAckStore();
+                    } else {
+                        completeHandleFuture();
+                    }
+                })
+                .exceptionally(t -> {
+                    changeToErrorState();
+                    exceptionHandleFuture(t);
+                    return null;
+                });
     }
 
     private void initPendingAckStore() {