codelipenghui commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612370039
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
##########
@@ -694,60 +705,192 @@ public void close() throws IOException {
}
});
- List<CompletableFuture<Void>> asyncCloseFutures = new
ArrayList<>();
+ CompletableFuture<CompletableFuture<Void>>
cancellableDownstreamFutureReference = new CompletableFuture<>();
+ CompletableFuture<Void> shutdownFuture =
+
CompletableFuture.allOf(shutdownEventLoopGracefully(acceptorGroup),
+ shutdownEventLoopGracefully(workerGroup))
+ .handle((v, t) -> {
+ if (t != null) {
+ log.warn("Error shutting down event loops
gracefully", t);
+ } else {
+ log.info("Event loops shutdown
completed.");
+ }
+ return null;
+ })
+ .thenCompose(__ -> {
+ log.info("Continuing to second phase in
shutdown.");
- if (listenChannel != null && listenChannel.isOpen()) {
- asyncCloseFutures.add(closeChannel(listenChannel));
- }
+ List<CompletableFuture<Void>>
asyncCloseFutures = new ArrayList<>();
- if (listenChannelTls != null && listenChannelTls.isOpen()) {
- asyncCloseFutures.add(closeChannel(listenChannelTls));
- }
+ if (listenChannel != null &&
listenChannel.isOpen()) {
+
asyncCloseFutures.add(closeChannel(listenChannel));
+ }
+
+ if (listenChannelTls != null &&
listenChannelTls.isOpen()) {
+
asyncCloseFutures.add(closeChannel(listenChannelTls));
+ }
+
+ if (interceptor != null) {
+ interceptor.close();
+ interceptor = null;
+ }
- acceptorGroup.shutdownGracefully();
- workerGroup.shutdownGracefully();
+ try {
+ authenticationService.close();
+ } catch (IOException e) {
+ log.warn("Error in closing
authenticationService", e);
+ }
+ pulsarStats.close();
+
ClientCnxnAspect.removeListener(zkStatsListener);
+ ClientCnxnAspect.registerExecutor(null);
+ try {
+ delayedDeliveryTrackerFactory.close();
+ } catch (IOException e) {
+ log.warn("Error in closing
delayedDeliveryTrackerFactory", e);
+ }
- if (interceptor != null) {
- interceptor.close();
- interceptor = null;
+
asyncCloseFutures.add(GracefulExecutorServiceShutdownHandler
+ .shutdownGracefully(
+ (long)
(GRACEFUL_SHUTDOWN_TIMEOUT_RATIO_OF_TOTAL_TIMEOUT
+ *
pulsar.getConfiguration().getBrokerShutdownTimeoutMs()),
+ statsUpdater,
+ inactivityMonitor,
+ messageExpiryMonitor,
+ compactionMonitor,
+ messagePublishBufferMonitor,
+ consumedLedgersMonitor,
+ backlogQuotaChecker,
+ topicOrderedExecutor,
+ topicPublishRateLimiterMonitor,
+
brokerPublishRateLimiterMonitor,
+ deduplicationSnapshotMonitor));
+
+ CompletableFuture<Void> combined =
+
FutureUtil.waitForAllAndSupportCancel(asyncCloseFutures);
+
cancellableDownstreamFutureReference.complete(combined);
+ combined.handle((v, t) -> {
+ if (t == null) {
+ log.info("Broker service completely
shut down");
+ } else {
+ if (t instanceof
CancellationException) {
+ log.warn("Broker service didn't
complete gracefully. "
+ + "Terminating Broker
service.");
+ } else {
+ log.warn("Broker service shut down
completed with exception", t);
+ }
+ }
+ return null;
+ });
+ return combined;
+ });
+ FutureUtil.whenCancelledOrTimedOut(shutdownFuture, () ->
cancellableDownstreamFutureReference
+ .thenAccept(future -> future.cancel(false)));
+ return shutdownFuture;
+ } catch (Exception e) {
+ return FutureUtil.failedFuture(e);
+ }
+ }
+
+ CompletableFuture<Void> shutdownEventLoopGracefully(EventLoopGroup
eventLoopGroup) {
+ long brokerShutdownTimeoutMs =
pulsar.getConfiguration().getBrokerShutdownTimeoutMs();
+ long quietPeriod = Math.min((long) (
+ GRACEFUL_SHUTDOWN_QUIET_PERIOD_RATIO_OF_TOTAL_TIMEOUT *
brokerShutdownTimeoutMs),
+ GRACEFUL_SHUTDOWN_QUIET_PERIOD_MAX_MS);
+ long timeout = (long)
(GRACEFUL_SHUTDOWN_TIMEOUT_RATIO_OF_TOTAL_TIMEOUT * brokerShutdownTimeoutMs);
+ return NettyFutureUtil.toCompletableFutureVoid(
+ eventLoopGroup.shutdownGracefully(quietPeriod,
+ timeout, TimeUnit.MILLISECONDS));
+ }
+
+ @Slf4j
+ private static class GracefulExecutorServiceShutdownHandler {
+ private final ScheduledExecutorService shutdownScheduler =
Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()));
+ private final List<ExecutorService> executors;
+ private final CompletableFuture<Void> future;
+ private final long timeoutMs;
+
+ private GracefulExecutorServiceShutdownHandler(long timeoutMs,
ExecutorService... executorServices) {
+ this.timeoutMs = timeoutMs;
+ executors = Arrays.stream(executorServices)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ static CompletableFuture<Void> shutdownGracefully(long timeoutMs,
ExecutorService... executorServices) {
+ return new GracefulExecutorServiceShutdownHandler(timeoutMs,
executorServices).doShutdownGracefully();
+ }
+
+ private CompletableFuture<Void> doShutdownGracefully() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkCompletion();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs,
TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
}
+ return future;
+ }
- statsUpdater.shutdown();
- inactivityMonitor.shutdown();
- messageExpiryMonitor.shutdown();
- compactionMonitor.shutdown();
- messagePublishBufferMonitor.shutdown();
- consumedLedgersMonitor.shutdown();
- backlogQuotaChecker.shutdown();
- authenticationService.close();
- pulsarStats.close();
- ClientCnxnAspect.removeListener(zkStatsListener);
- ClientCnxnAspect.registerExecutor(null);
- topicOrderedExecutor.shutdown();
- delayedDeliveryTrackerFactory.close();
- if (topicPublishRateLimiterMonitor != null) {
- topicPublishRateLimiterMonitor.shutdown();
+ private void terminate() {
+ for (ExecutorService executor : executors) {
+ if (!executor.isTerminated()) {
+ log.info("Shutting down forcefully executor {}", executor);
+ for (Runnable runnable : executor.shutdownNow()) {
+ log.info("Execution in progress for runnable instance
of {}: {}", runnable.getClass(),
+ runnable);
+ }
+ }
}
- if (brokerPublishRateLimiterMonitor != null) {
- brokerPublishRateLimiterMonitor.shutdown();
+ shutdown();
+ }
+
+ private void shutdown() {
+ if (!shutdownScheduler.isShutdown()) {
+ log.info("Shutting down scheduler.");
+ shutdownScheduler.shutdown();
}
- if (deduplicationSnapshotMonitor != null) {
- deduplicationSnapshotMonitor.shutdown();
+ }
+
+ private void scheduleCheck() {
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler
+ .schedule(this::checkCompletion,
Math.max(timeoutMs / 100, 10), TimeUnit.MILLISECONDS);
Review comment:
Why need `timeoutMs / 100` here? Is it should be `timeoutMs`
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -308,10 +310,17 @@ public void close() throws PulsarServerException {
try {
closeAsync().get();
} catch (ExecutionException e) {
- if (e.getCause() instanceof PulsarServerException) {
- throw (PulsarServerException) e.getCause();
+ Throwable cause = e.getCause();
+ if (cause instanceof PulsarServerException) {
+ throw (PulsarServerException) cause;
+ } else if (cause instanceof TimeoutException) {
+ if (getConfiguration().getBrokerShutdownTimeoutMs() < 1000L) {
+ // ignore shutdown timeout when it's less than 1000ms (in
tests)
Review comment:
The behavior should be consistent between the Pulsar server and the
test? If users configure the shutdown timeout which < 1000L, what is the
expected behavior.
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java
##########
@@ -52,30 +49,15 @@ public void run() {
+ service.getSafeWebServiceAddress() + ", broker url=" +
service.getSafeBrokerServiceUrl());
}
- ExecutorService executor = Executors.newSingleThreadExecutor(new
DefaultThreadFactory("shutdown-thread"));
-
try {
- CompletableFuture<Void> future = new CompletableFuture<>();
-
- executor.execute(() -> {
- try {
- service.closeAsync().whenComplete((result, throwable) -> {
- if (throwable != null) {
- future.completeExceptionally(throwable);
- } else {
- future.complete(result);
- }
- });
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- });
-
-
future.get(service.getConfiguration().getBrokerShutdownTimeoutMs(),
TimeUnit.MILLISECONDS);
-
+ service.closeAsync().get();
Review comment:
Why not use the "shutdown-thread" here? And if we don't apply the
future timeout, does the `processTerminator.accept(0)` can works? The thread
might always block by the get method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
