[fix][broker] Do not use IO thread for consumerFlow in Shared subscription
@@ -211,7 +211,13 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
@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);
