[fix][broker] call ServerCnx#closeProducer from correct thread (apach…

…e#20747)
lhotari · Jul 7, 2023 · 0bac873 · 0bac873
1 parent c1f52f8
commit 0bac873
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1613,7 +1613,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ
             }
 
             producers.remove(producerId, producerFuture);
-        }).exceptionally(ex -> {
+        }).exceptionallyAsync(ex -> {
             if (ex.getCause() instanceof BrokerServiceException.TopicMigratedException) {
                 Optional<ClusterUrl> clusterURL = getMigratedClusterUrl(service.getPulsar());
                 if (clusterURL.isPresent()) {
@@ -1654,7 +1654,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ
                         BrokerServiceException.getClientErrorCode(ex), ex.getMessage());
             }
             return null;
-        });
+        }, ctx.executor());
 
         producerQueuedFuture.thenRun(() -> {
             // If the producer is queued waiting, we will get an immediate notification
@@ -2933,6 +2933,7 @@ protected void interceptCommand(BaseCommand command) throws InterceptException {
 
     @Override
     public void closeProducer(Producer producer) {
+        assert ctx.executor().inEventLoop();
         // removes producer-connection from map and send close command to producer
         safelyRemoveProducer(producer);
         if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {