[fix][broker] Consume the ResourceGroup dispatch quota

Signed-off-by: Zixuan Liu <[email protected]>
ascentstream · Mar 6, 2024 · 1c40913 · 1c40913
1 parent 46556df
commit 1c40913
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...roker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/...roker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -397,6 +397,8 @@ public void readEntriesComplete(List<Entry> entries, Object ctx) {
                 }
 
                 dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));
+                resourceGroupDispatchRateLimiter.ifPresent(
+                        rateLimiter -> rateLimiter.consumeDispatchQuota(1, entry.getLength()));
 
                 msgOut.recordEvent(headersAndPayload.readableBytes());
 
@@ -799,7 +801,8 @@ public void initializeDispatchRateLimiterIfNeeded() {
             if (resourceGroupName != null) {
                 ResourceGroup resourceGroup = resourceGroupService.resourceGroupGet(resourceGroupName);
                 if (resourceGroup != null) {
-                    resourceGroupDispatchRateLimiter = Optional.of(resourceGroup.getResourceGroupReplicationDispatchLimiter());
+                    resourceGroupDispatchRateLimiter = Optional.of(resourceGroup
+                            .getResourceGroupReplicationDispatchLimiter());
                 }
             } else {
                 if (resourceGroupDispatchRateLimiter.isPresent()) {