[pulsar] branch branch-2.8 updated: [fix][broker] Fix deadlock in broker after race condition in topic creation failure (#15570)

[mmerli]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.8 by this push:
     new 2356dd50d68 [fix][broker] Fix deadlock in broker after race condition 
in topic creation failure (#15570)
2356dd50d68 is described below

commit 2356dd50d68e7acbfd769c4b7f255dacca6a083a
Author: Matteo Merli <mme...@apache.org>
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
---
 .../main/java/org/apache/pulsar/broker/service/BrokerService.java | 8 ++++----
 1 file changed, 4 insertions(+), 4 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 8695ee67042..c45a0ccb6a4 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
@@ -1354,9 +1354,9 @@ public class BrokerService implements Closeable, 
ZooKeeperCacheListener<Policies
                                     if 
(topicFuture.isCompletedExceptionally()) {
                                         log.warn("{} future is already 
completed with failure {}, closing the topic",
                                                 topic, 
FutureUtil.getException(topicFuture));
-                                        
persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
+                                        
persistentTopic.stopReplProducers().whenCompleteAsync((v, exception) -> {
                                             topics.remove(topic, topicFuture);
-                                        });
+                                        }, executor());
                                     } else {
                                         addTopicToStatsMaps(topicName, 
persistentTopic);
                                         
topicFuture.complete(Optional.of(persistentTopic));
@@ -1366,10 +1366,10 @@ public class BrokerService implements Closeable, 
ZooKeeperCacheListener<Policies
                                             "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;
                                 });