Re: [PR] [improve][broker] Pulsar Rate Limiting Refactoring changes (PIP-322) [pulsar]

["onobc (via GitHub)" <gi...@apache.org>]

onobc commented on code in PR #21681:
URL: https://github.com/apache/pulsar/pull/21681#discussion_r1419209100


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -898,57 +888,35 @@ public long increasePublishLimitedTimes() {
             .register();
 
     @Override
-    public void checkTopicPublishThrottlingRate() {
-        this.topicPublishRateLimiter.checkPublishRate();
-    }
+    public void incrementPublishCount(Producer producer, int numOfMessages, long msgSizeInBytes) {
+        handlePublishThrottling(producer, numOfMessages, msgSizeInBytes);
 
-    @Override
-    public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
-        // increase topic publish rate limiter
-        this.topicPublishRateLimiter.incrementPublishCount(numOfMessages, msgSizeInBytes);
-        // increase broker publish rate limiter
-        getBrokerPublishRateLimiter().incrementPublishCount(numOfMessages, msgSizeInBytes);
         // increase counters
         bytesInCounter.add(msgSizeInBytes);
         msgInCounter.add(numOfMessages);
     }
 
-    @Override
-    public void resetTopicPublishCountAndEnableReadIfRequired() {
-        // broker rate not exceeded. and completed topic limiter reset.
-        if (!getBrokerPublishRateLimiter().isPublishRateExceeded() && topicPublishRateLimiter.resetPublishCount()) {
-            enableProducerReadForPublishRateLimiting();
+    private void handlePublishThrottling(Producer producer, int numOfMessages, long msgSizeInBytes) {
+        // consume tokens from rate limiters and possibly throttle the connection that published the message
+        // if it's publishing too fast. Each connection will be throttled lazily when they publish messages.
+        for (PublishRateLimiter rateLimiter : activeRateLimiters) {
+            rateLimiter.handlePublishThrottling(producer, numOfMessages, msgSizeInBytes);
         }
     }
 
-    public void updateDispatchRateLimiter() {
-    }
-
-    @Override
-    public void resetBrokerPublishCountAndEnableReadIfRequired(boolean doneBrokerReset) {
-        // topic rate not exceeded, and completed broker limiter reset.
-        if (!topicPublishRateLimiter.isPublishRateExceeded() && doneBrokerReset) {
-            enableProducerReadForPublishRateLimiting();
+    private void updateActiveRateLimiters() {
+        if (isResourceGroupRateLimitingEnabled()) {
+            activeRateLimiters =
+                    List.of(this.topicPublishRateLimiter, getBrokerPublishRateLimiter(), resourceGroupPublishLimiter);
+        } else {
+            activeRateLimiters = List.of(this.topicPublishRateLimiter, getBrokerPublishRateLimiter());
         }
+        activeRateLimiters = activeRateLimiters.stream().filter(Objects::nonNull).toList();

Review Comment:
   This is probably fine as-is, but it might be safer to work on a new local list and do the stream/filter and then set the `activeRateLimiters` to the new list at the very end



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org