rebase code for apache#12294

wangjialing218 · Oct 21, 2021 · 99a776f · 99a776f
1 parent 02ba0a9
commit 99a776f
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 4 deletions.
diff --git a/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -589,8 +589,7 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
         long permits = dispatchThrottlingOnBatchMessageEnabled ? totalEntries : totalMessagesSent;
         if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
             if (topic.getBrokerDispatchRateLimiter().isPresent()) {
-                topic.getBrokerDispatchRateLimiter().get().tryDispatchPermit(totalMessagesSent,
-                        totalBytesSent);
+                topic.getBrokerDispatchRateLimiter().get().tryDispatchPermit(permits, totalBytesSent);
             }
             if (topic.getDispatchRateLimiter().isPresent()) {
                 topic.getDispatchRateLimiter().get().tryDispatchPermit(permits, totalBytesSent);

diff --git a/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -218,8 +218,8 @@ protected void dispatchEntriesToConsumer(Consumer currentConsumer, List<Entry> e
                     // acquire message-dispatch permits for already delivered messages
                     if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
                         if (topic.getBrokerDispatchRateLimiter().isPresent()) {
-                            topic.getBrokerDispatchRateLimiter().get().tryDispatchPermit(
-                                    sendMessageInfo.getTotalMessages(), sendMessageInfo.getTotalBytes());
+                            topic.getBrokerDispatchRateLimiter().get().tryDispatchPermit(permits,
+                                    sendMessageInfo.getTotalBytes());
                         }
 
                         if (topic.getDispatchRateLimiter().isPresent()) {