PIP-3 : Introduce message-dispatch rate limiting #634

rdhabalia · 2017-08-03T06:04:22Z

Motivation

As discussed in PIP, adding message-rate limiting to throttle message-dispatching.

Modifications

add rate-limiting while dispatching message for shared and exclusive/failOver subscription.
added RateLimiter to achieve per-second rate-limiting and also faster compare to others.
admiApi to set msg-rate at namespace and cluster level

Result

Broker can throttle message-dispatching at topic level.

rdhabalia · 2017-08-08T00:36:52Z

@merlimat as per your suggestion in #402, I have added rate-limiting by msg-rate and byte-rate. I will update the doc once you get a chance to review it.

merlimat

Overall approach looks good to me

merlimat · 2017-08-08T18:11:09Z

conf/broker.conf

@@ -106,6 +106,14 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages   
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16

+# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default
+# message dispatch-throttling 
+dispatchRatePerTopicInMsg=0


Can we include "throttling" in the var name? Like dispatchThrottlingRatePerTopicInMsg

merlimat · 2017-08-08T18:18:06Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -245,6 +245,19 @@ protected Policies getNamespacePolicies(String property, String cluster, String
        }
    }

+    protected LocalPolicies getNamespaceLocalPolicies(String property, String cluster, String namespace) {


Shouldn't we configuring the throttling policies on the global zk?

At least from an API perspective, it should be very clear what is the scope of the configuration

Shouldn't we configuring the throttling policies on the global zk?

Actually, every cluster has different traffic for the namespace and we may want to control enabling or throttling-limit per cluster level. Therefore, kept it in local-policies and not in global-zk.

That's fine, but it can be quite confusing for a user given that other methods are changing the settings globally.

I think it could either be explicitely set for a particular cluster, or for all the clusters. In order to clarify the scope. Eg:

{ "rateLimit" : { "dispatchRatePerTopicInMsg" : 1000.0, "dispatchRatePerTopicInBytes" : 1000000.0, "clusters" : { "small-cluster" : { "dispatchRatePerTopicInMsg" : 10.0, "dispatchRatePerTopicInBytes" : 1000.0, }, "large-cluster" : { "dispatchRatePerTopicInMsg" : 1000000.0, "dispatchRatePerTopicInBytes" : 1000000000.0, } } } }

The advantage is that you can also see all the limit in a single place.

Ok. But I think we introduced local-policies to keep cluster-level policy information such as BundleData and it keeps global-policies clean.

Though BundleData is different. It is not configuration. It is written back, by the brokers themselves, into the local ZK because we want each cluster to be able to split the bundles independently.

For rate limiting, we are talking about proper configuration.

merlimat · 2017-08-08T18:19:56Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java

@@ -810,6 +811,42 @@ public void splitNamespaceBundle(@PathParam("property") String property, @PathPa
        }
    }

+    @PUT


Should this POST rather than PUT?

have updated with POST but, REST standard suggests PUT for resource-update and POST for create.

merlimat · 2017-08-08T18:36:07Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

+            multiLayerTopicsMap.forEach((namespace, bundle) -> {
+                bundle.forEach((name, topics) -> {
+                    topics.forEach((topicName, topic) -> {
+                        if (topic instanceof PersistentTopic) {


Can't we just iterate on the flat topics map?

merlimat · 2017-08-08T18:36:48Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -575,5 +578,31 @@ private void clearUnAckedMsgs(Consumer consumer) {
        subscription.addUnAckedMessages(-unaAckedMsgs);
    }

+    public static class SendMessageInfo {


How is this information used?

At dispatcher, we want to know how many messages/bytes we actually sent to manage the permits. Therefore, at here in consumer, we exactly know how many msgs/bytes is sent and return it to dispatcher. Earlier, consumer.sendMessages() was returning Pair<ChannelPromise, Integer> and we wanted to add one more variable so, created SendMessageInfo entity to combine all information.

merlimat · 2017-08-08T19:04:18Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+        } catch (Exception e) {
+            log.warn("Failed to get message-rate for {}", this.topic, e);
+        }
+        DispatchRate dispatchRate = brokerService.pulsar().getConfiguration().getDispatchRatePerTopicInMsg() > 0


Ideally, if both limits are set, both of them should be applied.

Actually, that could just be done inside DispatchRate class

Actually, that could just be done inside DispatchRate class

Actually, using admin-api, we always define DispatchRateType=msgRate/byteRate and throttling-value in DispatchRate class.
However, for dynamic configuration we can't store complex-type so, created two variables dispatchRatePerTopicInMsg and dispatchRatePerTopicInByte, therefore, we need this logic when we have value set for both configuration.

Ideally, if both limits are set, both of them should be applied.

This will require 2 RateLimiter objects for every topic which we use across all the subscriptions. Do you think considering both limits would be useful?

It would require 2 rate limiters only if both limits are set. Otherwise one of them, or both, will be null.

merlimat · 2017-08-08T19:05:50Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java

+    }
+
+    @Override
+    public int hashCode() {


Unless hashCode() is used in a very critical place, just use Objects.hashCode(var1, var2)

merlimat · 2017-08-08T19:06:26Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java

+    }
+
+    @Override
+    public boolean equals(Object obj) {


return Objects.equals(a, obj.a) && Objects.equals(b, obj.b)

merlimat · 2017-08-08T19:07:08Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java

+    }
+
+    @Override
+    public String toString() {


return Objects.toString(this);

😄

merlimat · 2017-08-08T19:08:00Z

pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java

+ * acquire 10 permits at any time with in that 1 second.
+ *
+ */
+public class RateLimiter {


How is this class different from Guava rate limiter?

I have started with Guava-RateLimiter but ended up with custom one due to 2 main reasons:
(1) achieve per-second rate-limiting and (2) custom is faster compare to guava. I have added both Explanation here

Ok, I found out the gist with the explanation and it makes sense. Can you include the full explanation here as well? Also, would it make sense to have the 1sec period to be configurable as well? Eg: only rate limit if it's exceeding the rate for 10 seconds instead of 1.

read messages based on available-message-rate-permits add admin command add message-rate + byte-rate options for dispatch-rate limiting policies on global-zk + combined msg/byte rate + reset rateLimiter-timeUnit

rdhabalia · 2017-08-10T17:19:37Z

@merlimat I addressed all the comments.

merlimat

Looks good. Just few final comments

merlimat · 2017-08-10T17:23:42Z

pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java

+
+    }
+
+    public synchronized void shutdown() {


Can you implement AutoCloseable and change this into close() for consistency

merlimat · 2017-08-10T17:25:09Z

conf/broker.conf

@@ -106,6 +106,14 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages   
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16

+# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default


This should say the "messages per second" rate

merlimat · 2017-08-10T17:25:37Z

conf/broker.conf

+# message dispatch-throttling 
+dispatchThrottlingRatePerTopicInMsg=0
+
+# Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling


Same here, make it clear the meaning is bytes/sec

merlimat · 2017-08-10T20:53:17Z

...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java

+            // throttle only if: (1) cursor is not active bcz active-cursor reads message from cache rather from
+            // bookkeeper (2) if topic has reached message-rate threshold: then schedule the read after
+            // MESSAGE_RATE_BACKOFF_MS
+            if (!cursor.isActive()) {


Eventually, we could later add a config option to also rate limit the delivery from cache (that might help for topics with many subscriptions)

Eventually, we could later add a config option to also rate limit the delivery from cache (that might help for topics with many subscriptions)

yes, we also need this configuration to bound actual number of dispatch-messages, it is also helpful to throttle n/w bandwidth consumption by each client. I will create a separate PR to introduce this config.

merlimat · 2017-08-11T14:21:35Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java

+                dispatchRate);
+        validateSuperUserAccess();
+
+        if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {


It is not necessary to check the cluster here, change will be stored in global-zk and the right cluster will pick it up.

merlimat · 2017-08-11T18:22:49Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java

        CompletableFuture<Optional<LocalPolicies>> future = new CompletableFuture<>();
+        if (path == null || !path.startsWith(LOCAL_POLICIES_ROOT)) {


Is this still needed since we're not using local policies anymore?

actually, this method returns CompletableFuture<> so, we should fail the future rather throwing runtime exception.

merlimat

👍

rdhabalia added type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages type/feature The PR added a new feature or issue requested a new feature labels Aug 3, 2017

rdhabalia added this to the 1.20.0-incubating milestone Aug 3, 2017

rdhabalia self-assigned this Aug 3, 2017

rdhabalia changed the title ~~Introduce message-dispatch rate limiting~~ PIP-3 : Introduce message-dispatch rate limiting Aug 3, 2017

rdhabalia force-pushed the dispatch branch from 75753e8 to 07cfe88 Compare August 8, 2017 00:35

rdhabalia requested a review from merlimat August 8, 2017 00:36

rdhabalia force-pushed the dispatch branch 2 times, most recently from 6100afd to 1c1c031 Compare August 8, 2017 06:22

merlimat reviewed Aug 8, 2017

View reviewed changes

rdhabalia force-pushed the dispatch branch 7 times, most recently from dbd93c9 to bda64ac Compare August 10, 2017 04:51

Introduce message-dispatch rate limiting

65d7aaf

read messages based on available-message-rate-permits add admin command add message-rate + byte-rate options for dispatch-rate limiting policies on global-zk + combined msg/byte rate + reset rateLimiter-timeUnit

rdhabalia force-pushed the dispatch branch from bda64ac to 65d7aaf Compare August 10, 2017 05:50

merlimat reviewed Aug 11, 2017

View reviewed changes

address comment

408ebc0

merlimat approved these changes Aug 12, 2017

View reviewed changes

rdhabalia merged commit f03a8b7 into apache:master Aug 14, 2017

rdhabalia mentioned this pull request Aug 18, 2017

Rate limiting in the dispatcher #402

Closed

rdhabalia mentioned this pull request Sep 11, 2017

Add configuration to enable throttling for non-backlog consumers #754

Merged

rdhabalia deleted the dispatch branch October 6, 2021 22:49

rdhabalia mentioned this pull request Oct 7, 2021

[pulsar-broker] Support configuration to rate-limit dispatching on batch message #12294

Merged

                       }
                   }
+                  @PUT

		CompletableFuture<Optional<LocalPolicies>> future = new CompletableFuture<>();
		if (path == null \|\| !path.startsWith(LOCAL_POLICIES_ROOT)) {

PIP-3 : Introduce message-dispatch rate limiting #634

PIP-3 : Introduce message-dispatch rate limiting #634

Conversation

rdhabalia commented Aug 3, 2017

Motivation

Modifications

Result

rdhabalia commented Aug 8, 2017

merlimat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat Aug 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia commented Aug 10, 2017

merlimat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a comment

Choose a reason for hiding this comment

merlimat Aug 8, 2017 •

edited

Loading