[fix][broker] Do not use IO thread for consumerFlow in Shared subscri…

…ption (apache#16304)
gaozhangmin · Jul 14, 2022 · 163c360 · 163c360
1 parent 52b2208
commit 163c360
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -211,7 +211,13 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
     }
 
     @Override
-    public synchronized void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
+    public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
+        topic.getBrokerService().executor().execute(() -> {
+            internalConsumerFlow(consumer, additionalNumberOfMessages);
+        });
+    }
+
+    private synchronized void internalConsumerFlow(Consumer consumer, int additionalNumberOfMessages) {
         if (!consumerSet.contains(consumer)) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Ignoring flow control from disconnected consumer {}", name, consumer);