[fix][txn] Fix PendingAckHandleImpl when `pendingAckStoreProvider.checkInitializedBefore` failed
@@ -153,12 +153,19 @@ public class PendingAckHandleImpl extends PendingAckHandleState implements Pendi
this.pendingAckStoreProvider = this.persistentSubscription.getTopic()
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription)
+ .thenAccept(init -> {
if (init) {
initPendingAckStore();
} else {
completeHandleFuture();
}
+ })
+ .exceptionally(t -> {
+ changeToErrorState();
+ exceptionHandleFuture(t);
+ return null;
});
}
