You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/07/15 10:03:01 UTC
[GitHub] [pulsar] wangjialing218 opened a new pull request #11325: support broker level dispatch rate limiter
wangjialing218 opened a new pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325
### Motivation
Fixes #7720
Current we support broker level publish rate limiter.
This PR will add broker level dispatch rate limiter.
### Modifications
Add broker level dispatch rate limiter for msgs and bytes.
Refact calculateToRead() to reduce duplicate code.
### Verifying this change
Add testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling()
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-940574768
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] 315157973 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725630577
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ dispatchRateLimiter = pulsar.getBrokerService().getBrokerDispatchRateLimiter();
+ if (dispatchRateLimiter != null
+ && dispatchRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessagesEachTopic; i++) {
Review comment:
I suggest to stop the thread pool that refreshes the quota in rateLimiter at first, otherwise this unit test will often fail because we cannot predict the running of the thread
--
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
[GitHub] [pulsar] Anonymitaet commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-881094239
Thanks for your contribution.
The [PR template contains info about doc](https://github.com/apache/pulsar/blob/master/.github/PULL_REQUEST_TEMPLATE.md#documentation), which helps others know more about the changes. Can you provide doc info in future PR descriptions? Thanks
--
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
[GitHub] [pulsar] 315157973 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-938398881
The current reading process is like this:
1) Determine whether the threshold is exceeded
2) Enter the pending queue and prepare to read
3) The entry is successfully read, and the threshold count is deducted
There is a time difference between 1 and 3. When the Broker just restarts, if it is read concurrently, the problem in that PR will still exists
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725823952
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -317,6 +340,9 @@ public static DispatchRateImpl getPoliciesDispatchRate(final String cluster,
* @return
*/
public DispatchRate getPoliciesDispatchRate(BrokerService brokerService) {
+ if (type == Type.BROKER) {
Review comment:
Done
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-938487711
> The current reading process is like this:
>
> 1. Determine whether the threshold is exceeded
> 2. Enter the pending queue and prepare to read
> 3. The entry is successfully read, and the threshold count is deducted
>
> There is a time difference between 1 and 3. When the Broker just restarts, if it is read concurrently, the problem in that PR will still exists
The problem you mentioned may be solved in this PR #8611
--
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
[GitHub] [pulsar] 315157973 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-938602811
> > The current reading process is like this:
> >
> > 1. Determine whether the threshold is exceeded
> > 2. Enter the pending queue and prepare to read
> > 3. The entry is successfully read, and the threshold count is deducted
> >
> > There is a time difference between 1 and 3. When the Broker just restarts, if it is read concurrently, the problem in that PR will still exists
>
> The problem you mentioned may be solved in this PR #8611
This PR will only prevent the message from being sent to the consumer for a long time later, but it cannot solve the memory surge during concurrent reading when broker restart.
--
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
[GitHub] [pulsar] Anonymitaet commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-1016029547
As discussed w/ @wangjialing218, he will add the two parameters and corresponding descriptions to the [configuration - standalone](https://pulsar.apache.org/docs/en/next/reference-configuration/#standalone) page.
--
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
[GitHub] [pulsar] codelipenghui merged pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-941940123
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725740450
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -317,6 +340,9 @@ public static DispatchRateImpl getPoliciesDispatchRate(final String cluster,
* @return
*/
public DispatchRate getPoliciesDispatchRate(BrokerService brokerService) {
+ if (type == Type.BROKER) {
Review comment:
For broker dispatchRateLimiter `topicName` is null, add this judgment to avoid call `getPolicies` with null `topicName`.
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-939740110
@315157973 could you have a look again
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725836625
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ dispatchRateLimiter = pulsar.getBrokerService().getBrokerDispatchRateLimiter();
+ if (dispatchRateLimiter != null
+ && dispatchRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessagesEachTopic; i++) {
+ producer1.send(new byte[byteRate / 10]);
+ producer2.send(new byte[byteRate / 10]);
+ }
+ latch.await();
+ Assert.assertEquals(totalReceived.get(), numProducedMessages, 10);
+ long end = System.currentTimeMillis();
+ log.info("-- time to receive all messages: {} ", end - start);
+
+ // first 10 messages, which equals receiverQueueSize, will not wait.
+ Assert.assertTrue((end - start) >= 2000);
+
+ consumer1.close();
+ consumer2.close();
+ producer1.close();
+ producer2.close();
+ log.info("-- Exiting {} test --", methodName);
Review comment:
For topic dispatch rate limiter, we can configure from namespace level policies and topic level policies, and topic level policies will have a high priority. That's because the enforcement is done at the topic level.
For broker level, topic level and subscription level dispatch rate limiter, the enforcement is done at different level, so they have same priority and take effect together.
I have add a test, when all these dispatch rate limiter are configured, the minimum dispatch rate limiter should take effect in case of one consumer for one topic.
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-939199766
> This PR will only prevent the message from being sent to the consumer for a long time later, but it cannot solve the memory surge during concurrent reading when broker restart.
As current reading process, it's not easy to completely prevent the sudden increase entry read traffic when mutliply comsumers starting at same time.
With PR #8611 and this PR, the increase traffic will last only one second. This could reduce the influence of write traffic in case of #7720
--
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
[GitHub] [pulsar] Anonymitaet commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r670887824
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * verify broker level rate-limiting should throttle message-dispatching based on byte-rate
+ *
+ * <pre>
+ * 1. broker level dispatch-byte-rate = 1000 bytes/sec
+ * 2. start two consumers for two topics
+ * 3. send 15 msgs to each of the two topics: each msgs with 100 byte, total 3000 byte
+ * 4. it should take up to 2 second to receive all messages of the two topic
Review comment:
```suggestion
* 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
* 2. Start two consumers for two topics.
* 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
* 4. It should take up to 2 seconds to receive all messages of the two topics.
```
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * verify broker level rate-limiting should throttle message-dispatching based on byte-rate
Review comment:
```suggestion
* Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
```
do you mean this?
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-882169526
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] 315157973 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725754929
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -317,6 +340,9 @@ public static DispatchRateImpl getPoliciesDispatchRate(final String cluster,
* @return
*/
public DispatchRate getPoliciesDispatchRate(BrokerService brokerService) {
+ if (type == Type.BROKER) {
Review comment:
Should we use `topicName == null` instead of `type == Type.BROKER` ?
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-880568892
@codelipenghui @315157973 please have a look
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-1015128679
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-941818419
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725830410
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ dispatchRateLimiter = pulsar.getBrokerService().getBrokerDispatchRateLimiter();
+ if (dispatchRateLimiter != null
+ && dispatchRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessagesEachTopic; i++) {
Review comment:
Could you point out the thread which need to stop?
--
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
[GitHub] [pulsar] 315157973 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725831379
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -173,12 +188,20 @@ public void updateDispatchRate() {
}
updateDispatchRate(dispatchRate.get());
- log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
- }
+ if (type == Type.BROKER) {
+ log.info("configured broker message-dispatch rate {}", dispatchRate.get());
+ } else {
+ log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
+ }
+}
public static boolean isDispatchRateNeeded(BrokerService brokerService, Optional<Policies> policies,
String topicName, Type type) {
final ServiceConfiguration serviceConfig = brokerService.pulsar().getConfiguration();
+ if (type == Type.BROKER) {
Review comment:
OK
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725738547
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -173,12 +188,20 @@ public void updateDispatchRate() {
}
updateDispatchRate(dispatchRate.get());
- log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
- }
+ if (type == Type.BROKER) {
Review comment:
The existing log prints `topicName`. For broker dispatchRateLimiter `topicName` is null and makes log not friendly.
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-941818419
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] wangjialing218 removed a comment on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 removed a comment on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-1015128472
webServicePort
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-1015128472
webServicePort
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725828201
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -173,12 +188,20 @@ public void updateDispatchRate() {
}
updateDispatchRate(dispatchRate.get());
- log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
- }
+ if (type == Type.BROKER) {
+ log.info("configured broker message-dispatch rate {}", dispatchRate.get());
+ } else {
+ log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
+ }
+}
public static boolean isDispatchRateNeeded(BrokerService brokerService, Optional<Policies> policies,
String topicName, Type type) {
final ServiceConfiguration serviceConfig = brokerService.pulsar().getConfiguration();
+ if (type == Type.BROKER) {
Review comment:
Broker level and topic level rate dispath limiter have same priority, as long as subscription level rate limiter. Because the enforcement is done at different level.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
##########
@@ -238,6 +240,27 @@ public void resetCloseFuture() {
// noop
}
+ protected abstract void reScheduleRead();
+
+ protected boolean reachDispatchRateLimit(DispatchRateLimiter dispatchRateLimiter,
+ MutablePair<Integer, Long> calculateToRead) {
+ if (dispatchRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!dispatchRateLimiter.hasMessageDispatchPermit()) {
+ reScheduleRead();
+ return true;
+ } else {
+ // update messagesToRead according to available dispatch rate limit.
+ Pair<Integer, Long> calculateResult = computeReadLimits(calculateToRead.getLeft(),
+ (int) dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg(),
+ calculateToRead.getRight(),
+ dispatchRateLimiter.getAvailableDispatchRateLimitOnByte());
+ calculateToRead.setLeft(calculateResult.getLeft());
+ calculateToRead.setRight(calculateResult.getRight());
Review comment:
Done
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725828419
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
##########
@@ -296,55 +303,47 @@ public synchronized void readMoreEntries() {
messagesToRead = 1;
}
+ MutablePair<Integer, Long> calculateToRead = MutablePair.of(messagesToRead, bytesToRead);
Review comment:
Done
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-939846661
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-881159803
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-950449597
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] 315157973 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-940719508
Some unit tests seem to fail
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r746547184
##########
File path: conf/broker.conf
##########
@@ -327,6 +327,14 @@ brokerPublisherThrottlingMaxMessageRate=0
# (Disable byte rate limit with value 0)
brokerPublisherThrottlingMaxByteRate=0
+# Max Rate(in 1 seconds) of Message allowed to dispatch from a broker if broker dispatch rate limiting enabled
+# (Disable message rate limit with value 0)
+brokerDispatchThrottlingMaxMessageRate=0
Review comment:
```suggestion
dispatchThrottlingRateInMsg=0
```
##########
File path: conf/broker.conf
##########
@@ -327,6 +327,14 @@ brokerPublisherThrottlingMaxMessageRate=0
# (Disable byte rate limit with value 0)
brokerPublisherThrottlingMaxByteRate=0
+# Max Rate(in 1 seconds) of Message allowed to dispatch from a broker if broker dispatch rate limiting enabled
+# (Disable message rate limit with value 0)
+brokerDispatchThrottlingMaxMessageRate=0
+
+# Max Rate(in 1 seconds) of Byte allowed to dispatch from a broker if broker dispatch rate limiting enabled.
+# (Disable byte rate limit with value 0)
+brokerDispatchThrottlingMaxByteRate=0
Review comment:
```suggestion
dispatchThrottlingRateInByte=0
```
##########
File path: conf/broker.conf
##########
@@ -327,6 +327,14 @@ brokerPublisherThrottlingMaxMessageRate=0
# (Disable byte rate limit with value 0)
brokerPublisherThrottlingMaxByteRate=0
+# Max Rate(in 1 seconds) of Message allowed to dispatch from a broker if broker dispatch rate limiting enabled
+# (Disable message rate limit with value 0)
+brokerDispatchThrottlingMaxMessageRate=0
+
+# Max Rate(in 1 seconds) of Byte allowed to dispatch from a broker if broker dispatch rate limiting enabled.
+# (Disable byte rate limit with value 0)
+brokerDispatchThrottlingMaxByteRate=0
Review comment:
And we already have some `dispatchThrottling` related configurations such as `dispatchThrottlingRatePerTopicInByte`, it's better to put them together and follow the same name pattern
--
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
[GitHub] [pulsar] 315157973 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r724747615
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -317,6 +340,9 @@ public static DispatchRateImpl getPoliciesDispatchRate(final String cluster,
* @return
*/
public DispatchRate getPoliciesDispatchRate(BrokerService brokerService) {
+ if (type == Type.BROKER) {
Review comment:
According to the comment of this method, this judgment should not be put into this method.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -173,12 +188,20 @@ public void updateDispatchRate() {
}
updateDispatchRate(dispatchRate.get());
- log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
- }
+ if (type == Type.BROKER) {
Review comment:
The existing log has been able to print out the type and rate, why we need to output the log separately?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
##########
@@ -238,6 +240,27 @@ public void resetCloseFuture() {
// noop
}
+ protected abstract void reScheduleRead();
+
+ protected boolean reachDispatchRateLimit(DispatchRateLimiter dispatchRateLimiter,
+ MutablePair<Integer, Long> calculateToRead) {
+ if (dispatchRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!dispatchRateLimiter.hasMessageDispatchPermit()) {
+ reScheduleRead();
+ return true;
+ } else {
+ // update messagesToRead according to available dispatch rate limit.
+ Pair<Integer, Long> calculateResult = computeReadLimits(calculateToRead.getLeft(),
+ (int) dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg(),
+ calculateToRead.getRight(),
+ dispatchRateLimiter.getAvailableDispatchRateLimitOnByte());
+ calculateToRead.setLeft(calculateResult.getLeft());
+ calculateToRead.setRight(calculateResult.getRight());
Review comment:
It looks like it can be abstracted to another method, executed when reachDispatchRateLimit=false
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ dispatchRateLimiter = pulsar.getBrokerService().getBrokerDispatchRateLimiter();
+ if (dispatchRateLimiter != null
+ && dispatchRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessagesEachTopic; i++) {
+ producer1.send(new byte[byteRate / 10]);
+ producer2.send(new byte[byteRate / 10]);
+ }
+ latch.await();
+ Assert.assertEquals(totalReceived.get(), numProducedMessages, 10);
+ long end = System.currentTimeMillis();
+ log.info("-- time to receive all messages: {} ", end - start);
+
+ // first 10 messages, which equals receiverQueueSize, will not wait.
+ Assert.assertTrue((end - start) >= 2000);
+
+ consumer1.close();
+ consumer2.close();
+ producer1.close();
+ producer2.close();
+ log.info("-- Exiting {} test --", methodName);
Review comment:
Please add a unit test that test the priority of policies at different levels
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
Review comment:
we can use `awaitility`
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
##########
@@ -296,55 +303,47 @@ public synchronized void readMoreEntries() {
messagesToRead = 1;
}
+ MutablePair<Integer, Long> calculateToRead = MutablePair.of(messagesToRead, bytesToRead);
Review comment:
`calculateToRead ` is a high-frequency operation, I think we can avoid creating new objects, just pass two params to `reachDispatchRateLimit `
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ dispatchRateLimiter = pulsar.getBrokerService().getBrokerDispatchRateLimiter();
+ if (dispatchRateLimiter != null
+ && dispatchRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessagesEachTopic; i++) {
Review comment:
I suggest you stop the thread pool that refreshes the quota at first, otherwise this unit test will often fail because we cannot predict the running of the thread
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -173,12 +188,20 @@ public void updateDispatchRate() {
}
updateDispatchRate(dispatchRate.get());
- log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
- }
+ if (type == Type.BROKER) {
+ log.info("configured broker message-dispatch rate {}", dispatchRate.get());
+ } else {
+ log.info("[{}] configured {} message-dispatch rate at broker {}", this.topicName, type, dispatchRate.get());
+ }
+}
public static boolean isDispatchRateNeeded(BrokerService brokerService, Optional<Policies> policies,
String topicName, Type type) {
final ServiceConfiguration serviceConfig = brokerService.pulsar().getConfiguration();
+ if (type == Type.BROKER) {
Review comment:
Whether the priority of Broker level should be after Topic level？
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-938301114
@315157973 @codelipenghui @hangc0276 PTAL when you have time
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r725828580
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
##########
@@ -310,6 +310,102 @@ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionT
log.info("-- Exiting {} test --", methodName);
}
+ /**
+ * Verify whether the broker level rate-limiting is throttle message-dispatching based on byte-rate or not
+ *
+ * <pre>
+ * 1. Broker level dispatch-byte-rate is equal to 1000 bytes per second.
+ * 2. Start two consumers for two topics.
+ * 3. Send 15 msgs to each of the two topics. Each msgs with 100 bytes, thus 3000 bytes in total.
+ * 4. It should take up to 2 seconds to receive all messages of the two topics.
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 8000)
+ public void testBrokerBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace1 = "my-property/throttling_ns1";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/throttlingAll");
+ final String namespace2 = "my-property/throttling_ns2";
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace2 + "/throttlingAll");
+ final String subName = "my-subscriber-name-" + subscription;
+
+ final int byteRate = 1000;
+ admin.brokers().updateDynamicConfiguration("brokerDispatchThrottlingMaxByteRate", "" + byteRate);
+ admin.namespaces().createNamespace(namespace1, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(namespace2, Sets.newHashSet("test"));
+
+ final int numProducedMessagesEachTopic = 15;
+ final int numProducedMessages = numProducedMessagesEachTopic * 2;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic1", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName)
+ .receiverQueueSize(10)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in topic2", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1).create();
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+
+ boolean isMessageRateUpdate = false;
+ DispatchRateLimiter dispatchRateLimiter;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
Review comment:
Done
--
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
[GitHub] [pulsar] wangjialing218 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-942000268
@315157973 @codelipenghui could you have a check again
--
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
[GitHub] [pulsar] 315157973 commented on pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
315157973 commented on pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#issuecomment-938604800
We can support broker level dispatcher rate limiter, but this PR can not fix #7720
--
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
[GitHub] [pulsar] wangjialing218 commented on a change in pull request #11325: support broker level dispatch rate limiter
Posted by GitBox <gi...@apache.org>.
wangjialing218 commented on a change in pull request #11325:
URL: https://github.com/apache/pulsar/pull/11325#discussion_r747190903
##########
File path: conf/broker.conf
##########
@@ -327,6 +327,14 @@ brokerPublisherThrottlingMaxMessageRate=0
# (Disable byte rate limit with value 0)
brokerPublisherThrottlingMaxByteRate=0
+# Max Rate(in 1 seconds) of Message allowed to dispatch from a broker if broker dispatch rate limiting enabled
+# (Disable message rate limit with value 0)
+brokerDispatchThrottlingMaxMessageRate=0
+
+# Max Rate(in 1 seconds) of Byte allowed to dispatch from a broker if broker dispatch rate limiting enabled.
+# (Disable byte rate limit with value 0)
+brokerDispatchThrottlingMaxByteRate=0
Review comment:
Done
--
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