[Client] Continue shutdown of client when closing producers and consumers timeout
don't return exception to caller
@@ -750,24 +750,23 @@ public class PulsarClientImpl implements PulsarClient {
FutureUtil.addTimeoutHandling(combinedFuture, Duration.ofSeconds(CLOSE_TIMEOUT_SECONDS),
shutdownExecutor, () -> FutureUtil.createTimeoutException("Closing producers and consumers timed out.",
PulsarClientImpl.class, "closeAsync"));
- combinedFuture.whenComplete((__, t) -> new Thread(() -> {
+ combinedFuture.handle((__, t) -> {
if (t != null) {
log.error("Closing producers and consumers failed. Continuing with shutdown.", t);
}
+ new Thread(() -> {
shutdownExecutor.shutdownNow();
// All producers & consumers are now closed, we can stop the client safely
try {
shutdown();
- closeFuture.complete(null);
- state.set(State.Closed);
} catch (PulsarClientException e) {
- closeFuture.completeExceptionally(e);
+ log.error("Shutdown failed. Ignoring the exception.", e);
}
- }, "pulsar-client-shutdown-thread").start()).exceptionally(exception -> {
- closeFuture.completeExceptionally(exception);
+ state.set(State.Closed);
+ closeFuture.complete(null);
+ }, "pulsar-client-shutdown-thread").start();
return null;
});
-
return closeFuture;
}
