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 2022/04/24 11:34:29 UTC
[GitHub] [pulsar] Jason918 opened a new pull request, #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Jason918 opened a new pull request, #15295:
URL: https://github.com/apache/pulsar/pull/15295
### Motivation
See PIP-152 #15094
### Modifications
See #15094
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- org.apache.pulsar.broker.admin.TopicPoliciesTest#testSubscriptionLevelDispatchRate
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (**yes**)
- The admin cli options: (**yes**)
- Anything that affects deployment: (no)
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [ ] `doc`
Doc added in admin client api.
--
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] AnonHxy commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1118110577
LGTM
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r859454391
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java:
##########
@@ -71,6 +71,10 @@ public class TopicPolicies {
private Integer maxSubscriptionsPerTopic;
private DispatchRateImpl replicatorDispatchRate;
private SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+ /**
+ * Dispatch rate limiter setting for specific subscription.
+ */
+ private Map<String/*subscription*/, DispatchRateImpl> subscriptionLevelDispatchRateMap = new HashMap<>();
Review Comment:
Add `@Builder.Default` here.
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857358213
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -4396,6 +4397,60 @@ protected CompletableFuture<Void> internalRemoveSubscriptionDispatchRate(boolean
});
}
+ protected CompletableFuture<DispatchRate> internalGetSubscriptionLevelDispatchRate(String subName, boolean applied,
+ boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(otp -> {
+ DispatchRateImpl rate = otp.map(tp -> tp.getSubscriptionPolicies().get(subName))
+ .map(SubscriptionPolicies::getDispatchRate)
+ .orElse(null);
+ if (applied && rate == null) {
+ return internalGetSubscriptionDispatchRate(true, isGlobal);
+ } else {
+ return CompletableFuture.completedFuture(rate);
+ }
+ });
+ }
+
+ protected CompletableFuture<Void> internalSetSubscriptionLevelDispatchRate(String subName,
+ DispatchRateImpl dispatchRate,
+ boolean isGlobal) {
+ final DispatchRateImpl newDispatchRate = DispatchRateImpl.normalize(dispatchRate);
+ if (newDispatchRate == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
+ topicPolicies.setIsGlobal(isGlobal);
+ topicPolicies.getSubscriptionPolicies()
+ .computeIfAbsent(subName, k -> new SubscriptionPolicies())
+ .setDispatchRate(newDispatchRate);
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
+ });
+ }
+
+ protected CompletableFuture<Void> internalRemoveSubscriptionLevelDispatchRate(String subName, boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ if (!op.isPresent()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ TopicPolicies topicPolicies = op.get();
+ SubscriptionPolicies sp = topicPolicies.getSubscriptionPolicies().get(subName);
+ if (sp == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ sp.setDispatchRate(null);
+ if (sp.checkEmpty()) {
+ // cleanup empty SubscriptionPolicies
+ topicPolicies.getSubscriptionPolicies().remove(subName, sp);
+ }
+ topicPolicies.setIsGlobal(isGlobal);
Review Comment:
Got
--
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] Jason918 commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857224880
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java:
##########
@@ -71,6 +71,10 @@ public class TopicPolicies {
private Integer maxSubscriptionsPerTopic;
private DispatchRateImpl replicatorDispatchRate;
private SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+ /**
+ * Dispatch rate limiter setting for specific subscription.
+ */
+ private Map<String/*subscription*/, DispatchRateImpl> subscriptionLevelDispatchRateMap = new HashMap<>();
Review Comment:
> Do we need to introduce SubscriptionPolicies here? We might need to support other policies like backlog quota, TTL.
Good point.
--
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] Jason918 commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1110463322
@codelipenghui @nodece @complone PTAL
--
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 diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857213460
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java:
##########
@@ -71,6 +71,10 @@ public class TopicPolicies {
private Integer maxSubscriptionsPerTopic;
private DispatchRateImpl replicatorDispatchRate;
private SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+ /**
+ * Dispatch rate limiter setting for specific subscription.
+ */
+ private Map<String/*subscription*/, DispatchRateImpl> subscriptionLevelDispatchRateMap = new HashMap<>();
Review Comment:
Do we need to introduce SubscriptionPolicies here? We might need to support other policies like backlog quota, TTL.
--
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] Jason918 commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1112933892
/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] Jason918 merged pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 merged PR #15295:
URL: https://github.com/apache/pulsar/pull/15295
--
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] Technoboy- commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r864973623
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionPolicies.java:
##########
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.common.policies.data;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
+
+/**
+ * Subscription policies.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SubscriptionPolicies {
+ private DispatchRateImpl dispatchRate;
+
+ /**
+ * Check if this SubscriptionPolicies is empty. Empty SubscriptionPolicies can be auto removed from TopicPolicies.
+ * @return true if this SubscriptionPolicies is empty.
+ */
+ public boolean checkEmpty() {
Review Comment:
checkEmpty -> isEmpty ?
--
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] Jason918 commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857359630
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -1316,6 +1316,56 @@ public void testGetSetSubscriptionDispatchRate() throws Exception {
admin.topics().delete(topic, true);
}
+ @Test
+ public void testSubscriptionLevelDispatchRate() throws Exception{
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+
+ String subscriptionName = "testSubscriptionLevelDispatchRate";
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subscriptionName).topic(topic).subscribe();
+ AbstractTopic abstractTopic = (AbstractTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
+
+ DispatchRate topicLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(100)
+ .dispatchThrottlingRateInByte(1024)
+ .ratePeriodInSecond(1)
+ .build();
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, topicLevelRate);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic), topicLevelRate));
+
+ // topic level is set, applied value should be topic level setting.
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName, true),
+ topicLevelRate);
+ // subscription level setting is not set.
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
+
+ DispatchRate subLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(1000)
+ .dispatchThrottlingRateInByte(1024 * 1024)
+ .ratePeriodInSecond(1)
+ .build();
+
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, subscriptionName, subLevelRate);
+
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName), subLevelRate);
+
+ Awaitility.await().untilAsserted(() -> Assert.assertEquals(
+ abstractTopic.getSubscriptionDispatchRate(subscriptionName), subLevelRate));
+
+ DispatchRateLimiter dispatchRateLimiter = abstractTopic.getSubscription(subscriptionName)
+ .getDispatcher().getRateLimiter().get();
+ Assert.assertNotNull(dispatchRateLimiter);
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), subLevelRate.getDispatchThrottlingRateInByte());
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), subLevelRate.getDispatchThrottlingRateInMsg());
+
+ admin.topicPolicies().removeSubscriptionDispatchRate(topic, subscriptionName);
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
Review Comment:
fixed
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -1316,6 +1316,56 @@ public void testGetSetSubscriptionDispatchRate() throws Exception {
admin.topics().delete(topic, true);
}
+ @Test
+ public void testSubscriptionLevelDispatchRate() throws Exception{
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+
+ String subscriptionName = "testSubscriptionLevelDispatchRate";
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subscriptionName).topic(topic).subscribe();
+ AbstractTopic abstractTopic = (AbstractTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
+
+ DispatchRate topicLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(100)
+ .dispatchThrottlingRateInByte(1024)
+ .ratePeriodInSecond(1)
+ .build();
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, topicLevelRate);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic), topicLevelRate));
+
+ // topic level is set, applied value should be topic level setting.
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName, true),
+ topicLevelRate);
+ // subscription level setting is not set.
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
+
+ DispatchRate subLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(1000)
+ .dispatchThrottlingRateInByte(1024 * 1024)
+ .ratePeriodInSecond(1)
+ .build();
+
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, subscriptionName, subLevelRate);
+
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName), subLevelRate);
Review Comment:
fixed
--
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] github-actions[bot] commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1107822633
@Jason918:Thanks for providing doc info!
--
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 diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857214404
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java:
##########
@@ -71,6 +71,10 @@ public class TopicPolicies {
private Integer maxSubscriptionsPerTopic;
private DispatchRateImpl replicatorDispatchRate;
private SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+ /**
+ * Dispatch rate limiter setting for specific subscription.
+ */
+ private Map<String/*subscription*/, DispatchRateImpl> subscriptionLevelDispatchRateMap = new HashMap<>();
Review Comment:
And need to add `@Builder.Default` annotation? details to see https://github.com/apache/pulsar/pull/12009
--
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] complone commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
complone commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857265536
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,7 +160,13 @@ public SubscribeRate getSubscribeRate() {
return this.topicPolicies.getSubscribeRate().get();
}
- public DispatchRateImpl getSubscriptionDispatchRate() {
+ public DispatchRateImpl getSubscriptionDispatchRate(String subscriptionName) {
+ DispatchRateImpl rate = Optional.ofNullable(subscriptionPolicies.get(subscriptionName))
+ .map(SubscriptionPolicies::getDispatchRate).orElse(null);
+ rate = DispatchRateImpl.normalize(rate);
Review Comment:
Whether it should be judged here that ```rate``` is not empty
--
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] Jason918 commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1119393155
/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] Jason918 commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1109417036
/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] Jason918 closed pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 closed pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
URL: https://github.com/apache/pulsar/pull/15295
--
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] complone commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
complone commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1118088240
LGTM
--
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] Jason918 commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r865520805
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionPolicies.java:
##########
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.common.policies.data;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
+
+/**
+ * Subscription policies.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SubscriptionPolicies {
+ private DispatchRateImpl dispatchRate;
+
+ /**
+ * Check if this SubscriptionPolicies is empty. Empty SubscriptionPolicies can be auto removed from TopicPolicies.
+ * @return true if this SubscriptionPolicies is empty.
+ */
+ public boolean checkEmpty() {
Review Comment:
Some serialization lib would use `isXXX` method as boolean properties.
I prefer using `checkXXX` to indicate it's an action, not a property.
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857355458
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -1316,6 +1316,56 @@ public void testGetSetSubscriptionDispatchRate() throws Exception {
admin.topics().delete(topic, true);
}
+ @Test
+ public void testSubscriptionLevelDispatchRate() throws Exception{
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+
+ String subscriptionName = "testSubscriptionLevelDispatchRate";
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subscriptionName).topic(topic).subscribe();
+ AbstractTopic abstractTopic = (AbstractTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
+
+ DispatchRate topicLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(100)
+ .dispatchThrottlingRateInByte(1024)
+ .ratePeriodInSecond(1)
+ .build();
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, topicLevelRate);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic), topicLevelRate));
+
+ // topic level is set, applied value should be topic level setting.
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName, true),
+ topicLevelRate);
+ // subscription level setting is not set.
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
+
+ DispatchRate subLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(1000)
+ .dispatchThrottlingRateInByte(1024 * 1024)
+ .ratePeriodInSecond(1)
+ .build();
+
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, subscriptionName, subLevelRate);
+
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName), subLevelRate);
+
+ Awaitility.await().untilAsserted(() -> Assert.assertEquals(
+ abstractTopic.getSubscriptionDispatchRate(subscriptionName), subLevelRate));
+
+ DispatchRateLimiter dispatchRateLimiter = abstractTopic.getSubscription(subscriptionName)
+ .getDispatcher().getRateLimiter().get();
+ Assert.assertNotNull(dispatchRateLimiter);
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), subLevelRate.getDispatchThrottlingRateInByte());
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), subLevelRate.getDispatchThrottlingRateInMsg());
+
+ admin.topicPolicies().removeSubscriptionDispatchRate(topic, subscriptionName);
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
Review Comment:
Use `Awaitility`.
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857355102
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -1316,6 +1316,56 @@ public void testGetSetSubscriptionDispatchRate() throws Exception {
admin.topics().delete(topic, true);
}
+ @Test
+ public void testSubscriptionLevelDispatchRate() throws Exception{
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+
+ String subscriptionName = "testSubscriptionLevelDispatchRate";
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subscriptionName).topic(topic).subscribe();
+ AbstractTopic abstractTopic = (AbstractTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
+
+ DispatchRate topicLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(100)
+ .dispatchThrottlingRateInByte(1024)
+ .ratePeriodInSecond(1)
+ .build();
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, topicLevelRate);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic), topicLevelRate));
+
+ // topic level is set, applied value should be topic level setting.
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName, true),
+ topicLevelRate);
+ // subscription level setting is not set.
+ Assert.assertNull(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName));
+
+ DispatchRate subLevelRate = DispatchRate.builder()
+ .dispatchThrottlingRateInMsg(1000)
+ .dispatchThrottlingRateInByte(1024 * 1024)
+ .ratePeriodInSecond(1)
+ .build();
+
+ admin.topicPolicies().setSubscriptionDispatchRate(topic, subscriptionName, subLevelRate);
+
+ Assert.assertEquals(admin.topicPolicies().getSubscriptionDispatchRate(topic, subscriptionName), subLevelRate);
Review Comment:
Use `Awaitility`.
--
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] github-actions[bot] commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1107822533
@Jason918:Thanks for your contribution. For this PR, do we need to update docs?
(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-related info in this and 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] Jason918 commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857356575
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,7 +160,13 @@ public SubscribeRate getSubscribeRate() {
return this.topicPolicies.getSubscribeRate().get();
}
- public DispatchRateImpl getSubscriptionDispatchRate() {
+ public DispatchRateImpl getSubscriptionDispatchRate(String subscriptionName) {
+ DispatchRateImpl rate = Optional.ofNullable(subscriptionPolicies.get(subscriptionName))
+ .map(SubscriptionPolicies::getDispatchRate).orElse(null);
+ rate = DispatchRateImpl.normalize(rate);
Review Comment:
`DispatchRateImpl` don't have concept of `empty`. `normalize` can handle the case when rate is `null`.
--
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] AnonHxy commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r860516341
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,7 +160,13 @@ public SubscribeRate getSubscribeRate() {
return this.topicPolicies.getSubscribeRate().get();
}
- public DispatchRateImpl getSubscriptionDispatchRate() {
+ public DispatchRateImpl getSubscriptionDispatchRate(String subscriptionName) {
+ DispatchRateImpl rate = Optional.ofNullable(subscriptionPolicies.get(subscriptionName))
+ .map(SubscriptionPolicies::getDispatchRate).orElse(null);
+ rate = DispatchRateImpl.normalize(rate);
Review Comment:
This code can be written more convenient like this：
```
return Optional.ofNullable(subscriptionPolicies.get(subscriptionName))
.map(SubscriptionPolicies::getDispatchRate)
.map(DispatchRateImpl::normalize)
.orElse(this.topicPolicies.getSubscriptionDispatchRate().get());
```
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857363561
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java:
##########
@@ -71,6 +71,10 @@ public class TopicPolicies {
private Integer maxSubscriptionsPerTopic;
private DispatchRateImpl replicatorDispatchRate;
private SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+ /**
+ * Dispatch rate limiter setting for specific subscription.
+ */
+ private Map<String/*subscription*/, DispatchRateImpl> subscriptionLevelDispatchRateMap = new HashMap<>();
Review Comment:
You forget to add the `@Builder.Default` here.
--
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] nodece commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857337552
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -4396,6 +4397,60 @@ protected CompletableFuture<Void> internalRemoveSubscriptionDispatchRate(boolean
});
}
+ protected CompletableFuture<DispatchRate> internalGetSubscriptionLevelDispatchRate(String subName, boolean applied,
+ boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(otp -> {
+ DispatchRateImpl rate = otp.map(tp -> tp.getSubscriptionPolicies().get(subName))
+ .map(SubscriptionPolicies::getDispatchRate)
+ .orElse(null);
+ if (applied && rate == null) {
+ return internalGetSubscriptionDispatchRate(true, isGlobal);
+ } else {
+ return CompletableFuture.completedFuture(rate);
+ }
+ });
+ }
+
+ protected CompletableFuture<Void> internalSetSubscriptionLevelDispatchRate(String subName,
+ DispatchRateImpl dispatchRate,
+ boolean isGlobal) {
+ final DispatchRateImpl newDispatchRate = DispatchRateImpl.normalize(dispatchRate);
+ if (newDispatchRate == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
+ topicPolicies.setIsGlobal(isGlobal);
+ topicPolicies.getSubscriptionPolicies()
+ .computeIfAbsent(subName, k -> new SubscriptionPolicies())
+ .setDispatchRate(newDispatchRate);
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
+ });
+ }
+
+ protected CompletableFuture<Void> internalRemoveSubscriptionLevelDispatchRate(String subName, boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ if (!op.isPresent()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ TopicPolicies topicPolicies = op.get();
+ SubscriptionPolicies sp = topicPolicies.getSubscriptionPolicies().get(subName);
+ if (sp == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ sp.setDispatchRate(null);
+ if (sp.checkEmpty()) {
+ // cleanup empty SubscriptionPolicies
+ topicPolicies.getSubscriptionPolicies().remove(subName, sp);
+ }
+ topicPolicies.setIsGlobal(isGlobal);
Review Comment:
Why call the `topicPolicies.setIsGlobal(isGlobal)`?
--
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] complone commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
complone commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857265536
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,7 +160,13 @@ public SubscribeRate getSubscribeRate() {
return this.topicPolicies.getSubscribeRate().get();
}
- public DispatchRateImpl getSubscriptionDispatchRate() {
+ public DispatchRateImpl getSubscriptionDispatchRate(String subscriptionName) {
+ DispatchRateImpl rate = Optional.ofNullable(subscriptionPolicies.get(subscriptionName))
+ .map(SubscriptionPolicies::getDispatchRate).orElse(null);
+ rate = DispatchRateImpl.normalize(rate);
Review Comment:
Should it be judged here that ```rate``` is not empty, before calling the ```normalize``` function
--
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] Jason918 commented on a diff in pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#discussion_r857355240
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -4396,6 +4397,60 @@ protected CompletableFuture<Void> internalRemoveSubscriptionDispatchRate(boolean
});
}
+ protected CompletableFuture<DispatchRate> internalGetSubscriptionLevelDispatchRate(String subName, boolean applied,
+ boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(otp -> {
+ DispatchRateImpl rate = otp.map(tp -> tp.getSubscriptionPolicies().get(subName))
+ .map(SubscriptionPolicies::getDispatchRate)
+ .orElse(null);
+ if (applied && rate == null) {
+ return internalGetSubscriptionDispatchRate(true, isGlobal);
+ } else {
+ return CompletableFuture.completedFuture(rate);
+ }
+ });
+ }
+
+ protected CompletableFuture<Void> internalSetSubscriptionLevelDispatchRate(String subName,
+ DispatchRateImpl dispatchRate,
+ boolean isGlobal) {
+ final DispatchRateImpl newDispatchRate = DispatchRateImpl.normalize(dispatchRate);
+ if (newDispatchRate == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
+ topicPolicies.setIsGlobal(isGlobal);
+ topicPolicies.getSubscriptionPolicies()
+ .computeIfAbsent(subName, k -> new SubscriptionPolicies())
+ .setDispatchRate(newDispatchRate);
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
+ });
+ }
+
+ protected CompletableFuture<Void> internalRemoveSubscriptionLevelDispatchRate(String subName, boolean isGlobal) {
+ return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ .thenCompose(op -> {
+ if (!op.isPresent()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ TopicPolicies topicPolicies = op.get();
+ SubscriptionPolicies sp = topicPolicies.getSubscriptionPolicies().get(subName);
+ if (sp == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ sp.setDispatchRate(null);
+ if (sp.checkEmpty()) {
+ // cleanup empty SubscriptionPolicies
+ topicPolicies.getSubscriptionPolicies().remove(subName, sp);
+ }
+ topicPolicies.setIsGlobal(isGlobal);
Review Comment:
@nodece Just follow previous pattern. Maybe not necessary, but no harm.
--
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] AnonHxy commented on pull request #15295: [PIP-152] Support subscription level dispatch rate limiter setting.
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on PR #15295:
URL: https://github.com/apache/pulsar/pull/15295#issuecomment-1111783977
LGTM with a minor suggestion
--
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