This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new e345d24cf37 [fix][broker] Fix deadlock in broker after race condition
in topic creation failure (#15570)
e345d24cf37 is described below
commit e345d24cf37cfcd8433d1b32eabb7679774e375f
Author: Matteo Merli <[email protected]>
AuthorDate: Thu May 12 21:06:46 2022 -0700
[fix][broker] Fix deadlock in broker after race condition in topic creation
failure (#15570)
* Fix deadlock in broker after race condition in topic creation failure
* Fixed checkstyle
---
.../java/org/apache/pulsar/broker/service/BrokerService.java | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b7931be9b4b..ff3d248d592 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1423,9 +1423,10 @@ public class BrokerService implements Closeable {
if
(topicFuture.isCompletedExceptionally()) {
log.warn("{} future is already
completed with failure {}, closing the"
+ " topic", topic,
FutureUtil.getException(topicFuture));
-
persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
- topics.remove(topic,
topicFuture);
- });
+
persistentTopic.stopReplProducers()
+ .whenCompleteAsync((v,
exception) -> {
+
topics.remove(topic, topicFuture);
+ }, executor());
} else {
addTopicToStatsMaps(topicName,
persistentTopic);
topicFuture.complete(Optional.of(persistentTopic));
@@ -1434,10 +1435,10 @@ public class BrokerService implements Closeable {
.exceptionally((ex) -> {
log.warn("Replication or dedup
check failed."
+ " Removing topic from
topics list {}, {}", topic, ex);
-
persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
+
persistentTopic.stopReplProducers().whenCompleteAsync((v, exception) -> {
topics.remove(topic,
topicFuture);
topicFuture.completeExceptionally(ex);
- });
+ }, executor());
return null;
});
} catch (PulsarServerException e) {
