Fix OpBase.callback is not called in failPendingRequest
@@ -125,7 +125,6 @@ public class TransactionMetaStoreHandler extends HandlerState
if (getState() == State.Closing || getState() == State.Closed) {
setState(State.Closed);
failPendingRequest();
- this.pendingRequests.clear();
return;
}
@@ -175,9 +174,8 @@ public class TransactionMetaStoreHandler extends HandlerState
}
private void failPendingRequest() {
- internalPinnedExecutor.execute(() -> {
- pendingRequests.keys().forEach(k -> {
- OpBase<?> op = pendingRequests.remove(k);
+ // this method is executed in internalPinnedExecutor.
+ pendingRequests.forEach((k, op) -> {
if (op != null && !op.callback.isDone()) {
op.callback.completeExceptionally(new PulsarClientException.AlreadyClosedException(
"Could not get response from transaction meta store when "
@@ -185,7 +183,7 @@ public class TransactionMetaStoreHandler extends HandlerState
onResponse(op);
}
});
- });
+ this.pendingRequests.clear();
}
public CompletableFuture<TxnID> newTransactionAsync(long timeout, TimeUnit unit) {
