You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/11/10 11:08:12 UTC
(pulsar) branch branch-3.0 updated: [fix][broker] Fix PulsarService/BrokerService shutdown when brokerShutdownTimeoutMs=0 (#21496)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 18f577045ee [fix][broker] Fix PulsarService/BrokerService shutdown when brokerShutdownTimeoutMs=0 (#21496)
18f577045ee is described below
commit 18f577045ee96c6c7c53b8ba206b5d7b01f3804a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Nov 1 20:38:48 2023 +0200
[fix][broker] Fix PulsarService/BrokerService shutdown when brokerShutdownTimeoutMs=0 (#21496)
---
.../src/main/java/org/apache/pulsar/broker/PulsarService.java | 10 +++++++---
.../java/org/apache/pulsar/broker/service/BrokerService.java | 9 ++++++++-
2 files changed, 15 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 87e1a17b9b0..edcc83df2b8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -632,14 +632,18 @@ public class PulsarService implements AutoCloseable, ShutdownService {
}
private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ long brokerShutdownTimeoutMs = getConfiguration().getBrokerShutdownTimeoutMs();
+ if (brokerShutdownTimeoutMs <= 0) {
+ return future;
+ }
ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
new ExecutorProvider.ExtendedThreadFactory(getClass().getSimpleName() + "-shutdown"));
FutureUtil.addTimeoutHandling(future,
- Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ Duration.ofMillis(brokerShutdownTimeoutMs),
shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
future.handle((v, t) -> {
- if (t != null && getConfiguration().getBrokerShutdownTimeoutMs() > 0) {
- LOG.info("Shutdown timed out after {} ms", getConfiguration().getBrokerShutdownTimeoutMs());
+ if (t instanceof TimeoutException) {
+ LOG.info("Shutdown timed out after {} ms", brokerShutdownTimeoutMs);
LOG.info(ThreadDumpUtil.buildThreadDiagnosticString());
}
// shutdown the shutdown executor
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b5bd7bfaa3d..6e9be5d8faf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -825,6 +825,7 @@ public class BrokerService implements Closeable {
for (EventLoopGroup group : protocolHandlersWorkerGroups) {
shutdownEventLoops.add(shutdownEventLoopGracefully(group));
}
+
CompletableFuture<Void> shutdownFuture =
CompletableFuture.allOf(shutdownEventLoops.toArray(new CompletableFuture[0]))
.handle((v, t) -> {
@@ -835,7 +836,7 @@ public class BrokerService implements Closeable {
}
return null;
})
- .thenCompose(__ -> {
+ .thenComposeAsync(__ -> {
log.info("Continuing to second phase in shutdown.");
List<CompletableFuture<Void>> asyncCloseFutures = new ArrayList<>();
@@ -899,6 +900,12 @@ public class BrokerService implements Closeable {
return null;
});
return combined;
+ }, runnable -> {
+ // run the 2nd phase of the shutdown in a separate thread
+ Thread thread = new Thread(runnable);
+ thread.setName("BrokerService-shutdown-phase2");
+ thread.setDaemon(false);
+ thread.start();
});
FutureUtil.whenCancelledOrTimedOut(shutdownFuture, () -> cancellableDownstreamFutureReference
.thenAccept(future -> future.cancel(false)));