[Issue 11581][broker] feat:pass the executor to RateLimiter in Resour…

…ceGroupPublishLimiter (#11582) ### background Fixes #11581 ### Motivation we need to pass the executor to RateLimiter in ResourceGroupPublishLimiter ### Modifications pass the executor to RateLimiter in ResourceGroupPublishLimiter
apache · Aug 21, 2021 · f7957a5 · f7957a5
1 parent d10100f
commit f7957a5
Showing 1 changed file with 8 additions and 8 deletions.
diff --git a/...ker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java b/...ker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java
@@ -95,23 +95,23 @@ public void update(ResourceGroup resourceGroup) {
                 this.publishMaxMessageRate = Math.max(resourceGroup.getPublishRateInMsgs(), 0);
                 this.publishMaxByteRate = Math.max(resourceGroup.getPublishRateInBytes(), 0);
                 if (this.publishMaxMessageRate > 0) {
-                    // TODO: pass the executor
                     publishRateLimiterOnMessage = RateLimiter.builder()
+                            .scheduledExecutorService(scheduledExecutorService)
                             .permits(publishMaxMessageRate)
                             .rateTime(1L)
                             .timeUnit(TimeUnit.SECONDS)
                             .rateLimitFunction(this::apply)
                             .build();
                 }
                 if (this.publishMaxByteRate > 0) {
-                    // TODO: pass the executor
                     publishRateLimiterOnByte =
-                            RateLimiter.builder()
-                                    .permits(publishMaxByteRate)
-                                    .rateTime(1L)
-                                    .timeUnit(TimeUnit.SECONDS)
-                                    .rateLimitFunction(this::apply)
-                                    .build();
+                    RateLimiter.builder()
+                            .scheduledExecutorService(scheduledExecutorService)
+                            .permits(publishMaxByteRate)
+                            .rateTime(1L)
+                            .timeUnit(TimeUnit.SECONDS)
+                            .rateLimitFunction(this::apply)
+                            .build();
                 }
             } else {
                 this.publishMaxMessageRate = 0;