You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/11/23 19:41:35 UTC
(pulsar) 01/03: [fix][broker] Continue closing even when executor is shut down (#22599)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e2bd42311f7d3a5f895f63426fc3538aa6c54e05
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Apr 26 23:16:54 2024 +0300
[fix][broker] Continue closing even when executor is shut down (#22599)
(cherry picked from commit f8f256cfbdcd780c81442dc5566b6ed071141645)
---
.../pulsar/broker/service/persistent/PersistentTopic.java | 10 +++++++++-
1 file changed, 9 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 337fad5e6b6..4828857878b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -45,6 +45,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -1452,7 +1453,14 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
FutureUtil.waitForAll(futures).thenRunAsync(() -> {
closeClientFuture.complete(null);
- }, getOrderedExecutor()).exceptionally(ex -> {
+ }, command -> {
+ try {
+ getOrderedExecutor().execute(command);
+ } catch (RejectedExecutionException e) {
+ // executor has been shut down, execute in current thread
+ command.run();
+ }
+ }).exceptionally(ex -> {
log.error("[{}] Error closing clients", topic, ex);
alreadyUnFenced.set(true);
unfenceTopicToResume();