You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/04/12 14:54:40 UTC
[GitHub] [pulsar] lhotari opened a new pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
lhotari opened a new pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199
### Motivation
Broker shutdown isn't currently graceful although the intention for it is to be graceful.
Another problem in tests is that broker shutdown doesn't shutdown synchronously.
These 2 problems are addressed in this PR.
### Modifications
- Goal of changes: shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
- wait for event loops getting closed before continuing to close other services in broker shutdown
- configure event loop shutdown parameters since the default shutdown timeout is 15 seconds after a 2 second idle time.
- close executor services forcefully if the broker shutdown times out
- use shutdown timeout of 100 milliseconds in tests
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612555706
##########
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 reason to do this choice is to simplify the configuration for tests. This PR sets the brokerShutdownTimeMs to 100 ms in tests. If the timeout exception wouldn't be ignored, special handling would be needed in all tests to catch the TimeoutException and ignore it. I'm expecting that it's a justified tradeoff to have different behavior if the user goes and configured shutdown timeout to less than 1000 ms. WDYT?
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819883315
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819059882
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818655278
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818584475
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818395250
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819929946
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820544801
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612167254
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/CompletableFutureCancellationHandler.java
##########
@@ -41,10 +41,16 @@
* any "downstream" dependent futures. A cancellation or timeout that happens in any "upstream"
* future will get handled.
*/
-class CompletableFutureCancellationHandler {
- private volatile boolean cancelled;
+public class CompletableFutureCancellationHandler {
+ private enum CompletionStatus {
+ PENDING,
+ CANCELLED,
+ DONE
+ }
+ private volatile CompletionStatus completionStatus = CompletionStatus.PENDING;
private volatile Runnable cancelAction;
private final AtomicBoolean cancelHandled = new AtomicBoolean();
+ private boolean attached;
Review comment:
it should be fine since `attached` is used only in `attachToFuture` method and that method is synchronized. The reason why I added the "attached" field is to prevent misuse of `CompletableFutureCancellationHandler` class. It is designed to be "attached" to a single CompletableFuture and it cannot be reused.
--
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:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820528672
@merlimat to review
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819198134
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614694023
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
oh sorry, I didn't look at the full context of the code where this was. In this case, I'll simply use `shutdownNow` to terminate the scheduler used to schedule the timeout handling for the CompletableFuture.
--
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:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612916574
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/CompletableFutureCancellationHandler.java
##########
@@ -41,10 +41,16 @@
* any "downstream" dependent futures. A cancellation or timeout that happens in any "upstream"
* future will get handled.
*/
-class CompletableFutureCancellationHandler {
- private volatile boolean cancelled;
+public class CompletableFutureCancellationHandler {
+ private enum CompletionStatus {
+ PENDING,
+ CANCELLED,
+ DONE
+ }
+ private volatile CompletionStatus completionStatus = CompletionStatus.PENDING;
private volatile Runnable cancelAction;
private final AtomicBoolean cancelHandled = new AtomicBoolean();
+ private boolean attached;
Review comment:
@lhotari - you're right, it is correct as written. I had a small misunderstanding about the guarantees of the `synchronized` keyword. (When the method's lock is released, the updates to `attached` will be flushed to main memory and then when the method's lock is re-acquired, another thread will have a consistent view of the variable.) Sorry for the confusion!
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r613931149
##########
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:
I guess one alternative is to use the timeout value of 0 in tests and ignore the timeout exceptions when timeout is 0. That could be less surprising than having the 1000ms boundary.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818395250
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819059882
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819225712
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614691047
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
+ }
+ }
+ }
+ shutdown();
+ }
+
+ private void shutdown() {
+ if (!shutdownScheduler.isShutdown()) {
+ log.info("Shutting down scheduler.");
+ shutdownScheduler.shutdown();
Review comment:
This is now handled in the rewritten solution.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819288419
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614566281
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
Review comment:
good catch. I'll remove this since it's a leftover of some earlier logic.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819130696
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819198134
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819326437
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r613949002
##########
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:
But even if we ignore the timeout under 1000ms, the test still can occur the TimeoutException? Most tests are extends from the `MockedPulsarServiceBaseTest`, can we deal with this in the `MockedPulsarServiceBaseTest`?
##########
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:
I see.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818655278
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818319257
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819367818
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818442949
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819803044
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818012062
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818442949
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819454149
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820026053
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818542201
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819803044
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819883315
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818653737
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614600285
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
+ }
+ }
+ }
+ shutdown();
+ }
+
+ private void shutdown() {
+ if (!shutdownScheduler.isShutdown()) {
+ log.info("Shutting down scheduler.");
+ shutdownScheduler.shutdown();
Review comment:
I'll rewrite the logic to use a `java.lang.Thread` instead of `ScheduledExecutorService`.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-821048730
> The approach LGTM. Just a couple of questions on the await termination
@merlimat Thank you for your review and good points about await termination.
Please ignore some of the replies I made since I didn't look at the full context of the code when replying.
I revisited the code after another check at your review comments.
I have rewritten the solution to use `.awaitTermination` and replaced the previous solution that used a ScheduledExecutorService. Please take another look.
--
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:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r613949893
##########
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:
> I guess one alternative is to use the timeout value of 0 in tests and ignore the timeout exceptions when timeout is 0. That could be less surprising than having the 1000ms boundary
Sound good to me.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614690537
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
I have now rewritten the logic to use `awaitTermination`. @merlimat please take another look
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
Review comment:
I have now rewritten the logic to use `awaitTermination`.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820302083
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819949208
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818616759
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820343228
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819130696
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819552431
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818233768
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819949208
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819454149
--
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:
users@infra.apache.org
[GitHub] [pulsar] merlimat merged pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
merlimat merged pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818653737
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819258199
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612562476
##########
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:
I reverted the changes so it's back to the original logic. The reason I had made the changes is the fact that BrokerService now contains timeout handling as well. However this code calls the shutdown of PulsarService so there could be possibly some call that blocks and the original logic is safer.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818437087
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818012062
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818688151
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614240424
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
Review comment:
Isn't the shutdown already triggered when the executors are added to `GracefulExecutorServicesShutdown` ?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
Should we do `shutdownExecutor.shutdownDown() && shutdownExecutor.await()` here?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
Review comment:
Shouldn't we do a `executor.awaitTermination()` here to ensure the thread is really gone before we proceed? (Or at least we give it the chance to exit)
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
+ }
+ }
+ }
+ shutdown();
+ }
+
+ private void shutdown() {
+ if (!shutdownScheduler.isShutdown()) {
+ log.info("Shutting down scheduler.");
+ shutdownScheduler.shutdown();
Review comment:
Also for the scheduler thread, shouldn't we await for the thread termination?
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614565941
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
~~`shutdown` is called from the thread executing in the shutdownExecutor. Therefore calling `.shutdownDown` or `.await` aren't suitable. (It's not really needed either since there aren't any blocking methods that the shutdownExecutor calls.)~~
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818688151
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819367818
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612137732
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/CompletableFutureCancellationHandler.java
##########
@@ -41,10 +41,16 @@
* any "downstream" dependent futures. A cancellation or timeout that happens in any "upstream"
* future will get handled.
*/
-class CompletableFutureCancellationHandler {
- private volatile boolean cancelled;
+public class CompletableFutureCancellationHandler {
+ private enum CompletionStatus {
+ PENDING,
+ CANCELLED,
+ DONE
+ }
+ private volatile CompletionStatus completionStatus = CompletionStatus.PENDING;
private volatile Runnable cancelAction;
private final AtomicBoolean cancelHandled = new AtomicBoolean();
+ private boolean attached;
Review comment:
Do you think it is worth adding the `volatile` keyword to `attached`? I can see that `attachToFuture` is synchronized, but if the method were called from two separate threads in close succession, I think it could lead to a race condition.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819258199
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r612550100
##########
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:
It's intentional to use `Math.max((timeoutMs / 100, 10)` as the interval for checking if all executors have been terminated. If we'd use `timeoutMs`, we'd have to wait for the complete timeout duration until the check is made.
The check is fairly cheap, so it's fine to do the check fairly often. When using a timeout of 30000 ms, the check would be made every 300 milliseconds. Currently the minimum checking interval is 10 milliseconds.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818616759
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818584475
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820004562
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820004562
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818037978
This PRs continues the work started in PR #9308 .
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614599387
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
After thinking about it again, yes. I'll rewrite the logic to use `awaitTermination`.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614031191
##########
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:
I pushed changes to address this. @codelipenghui Please take another look.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-821192431
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614599717
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/GracefulExecutorServicesShutdownHandler.java
##########
@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Shuts down one or many {@link ExecutorService}s in a graceful way.
+ *
+ * The executors will be terminated forcefully after the timeout or when the future is cancelled.
+ *
+ * Designed to be used via the API in {@link GracefulExecutorServicesShutdown}
+ */
+@Slf4j
+class GracefulExecutorServicesShutdownHandler {
+ 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 final long shutdownStatusPollingInterval;
+
+ GracefulExecutorServicesShutdownHandler(long timeoutMs, List<ExecutorService> executorServices) {
+ this.timeoutMs = Math.max(timeoutMs, 1L);
+ this.shutdownStatusPollingInterval = Math.min(Math.max(timeoutMs / 100, 10), timeoutMs);
+ executors = executorServices.stream()
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ future = new CompletableFuture<>();
+ }
+
+ CompletableFuture<Void> startShutdownHandler() {
+ log.info("Shutting down {} executors.", executors.size());
+ executors.forEach(ExecutorService::shutdown);
+ FutureUtil.whenCancelledOrTimedOut(future, () -> {
+ terminate();
+ });
+ checkIfExecutorsHaveBeenShutdown();
+ if (!shutdownScheduler.isShutdown()) {
+ try {
+ shutdownScheduler.schedule(this::terminate, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // ignore
+ }
+ }
+ return future;
+ }
+
+ 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);
Review comment:
Yes, I'll rewrite the logic to use `awaitTermination`.
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614695134
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
Switched to use `shutdownNow`
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819683640
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819225712
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818319257
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818714132
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818542201
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a change in pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#discussion_r614565941
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -480,6 +487,20 @@ public void close() throws PulsarServerException {
}
}
+ private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
+ ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ FutureUtil.addTimeoutHandling(future,
+ Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
+ shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
+ future.handle((v, t) -> {
+ // shutdown the shutdown executor
+ shutdownExecutor.shutdown();
Review comment:
`shutdown` is called from the thread executing in the shutdownExecutor. Therefore calling `.shutdownDown` or `.await` aren't suitable. (It's not really needed either since there aren't any blocking methods that the shutdownExecutor calls.)
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818437087
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819683640
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-820081852
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818714132
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-818233768
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org
[GitHub] [pulsar] lhotari removed a comment on pull request #10199: Shutdown Broker gracefully, but forcefully after brokerShutdownTimeoutMs
Posted by GitBox <gi...@apache.org>.
lhotari removed a comment on pull request #10199:
URL: https://github.com/apache/pulsar/pull/10199#issuecomment-819929946
/pulsarbot run-failure-checks
--
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:
users@infra.apache.org