Skip to content

[broker] Restore clusterDispatchRate policy for compatibility #6176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sijie merged 3 commits into from
Feb 10, 2020
Merged

[broker] Restore clusterDispatchRate policy for compatibility #6176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6ee175b
Restore clusterDispatchRate policy for compatibility
Jan 30, 2020
2136876
Merge remote-tracking branch 'apache/master' into restore-cluster-dis…
sijie Feb 4, 2020
4c1cc09
Merge remote-tracking branch 'apache/master' into restore-cluster-dis…
sijie Feb 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -855,6 +855,7 @@ protected PublishRate internalGetPublishRate() {
}
}

@SuppressWarnings("deprecation")
protected void internalSetTopicDispatchRate(DispatchRate dispatchRate) {
log.info("[{}] Set namespace dispatch-rate {}/{}", clientAppId(), namespaceName, dispatchRate);
validateSuperUserAccess();
Expand All @@ -867,6 +868,7 @@ protected void internalSetTopicDispatchRate(DispatchRate dispatchRate) {
policiesNode = policiesCache().getWithStat(path).orElseThrow(
() -> new RestException(Status.NOT_FOUND, "Namespace " + namespaceName + " does not exist"));
policiesNode.getKey().topicDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);
policiesNode.getKey().clusterDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);

// Write back the new policies into zookeeper
globalZk().setData(path, jsonMapper().writeValueAsBytes(policiesNode.getKey()),
Expand All @@ -892,11 +894,15 @@ protected void internalSetTopicDispatchRate(DispatchRate dispatchRate) {
}
}

@SuppressWarnings("deprecation")
protected DispatchRate internalGetTopicDispatchRate() {
validateAdminAccessForTenant(namespaceName.getTenant());

Policies policies = getNamespacePolicies(namespaceName);
DispatchRate dispatchRate = policies.topicDispatchRate.get(pulsar().getConfiguration().getClusterName());
if (dispatchRate == null) {
dispatchRate = policies.clusterDispatchRate.get(pulsar().getConfiguration().getClusterName());
}
if (dispatchRate != null) {
return dispatchRate;
} else {
Expand Down
8 changes: 8 additions & 0 deletions ...broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,7 @@ public static boolean isDispatchRateNeeded(final ServiceConfiguration serviceCon
return true;
}

@SuppressWarnings("deprecation")
public void onPoliciesUpdate(Policies data) {
String cluster = brokerService.pulsar().getConfiguration().getClusterName();

Expand All @@ -194,6 +195,9 @@ public void onPoliciesUpdate(Policies data) {
switch (type) {
case TOPIC:
dispatchRate = data.topicDispatchRate.get(cluster);
if (dispatchRate == null) {
dispatchRate = data.clusterDispatchRate.get(cluster);
}
break;
case SUBSCRIPTION:
dispatchRate = data.subscriptionDispatchRate.get(cluster);
Expand All @@ -219,13 +223,17 @@ public void onPoliciesUpdate(Policies data) {
}
}

@SuppressWarnings("deprecation")
public static DispatchRate getPoliciesDispatchRate(final String cluster, Optional<Policies> policies, Type type) {
// return policy-dispatch rate only if it's enabled in policies
return policies.map(p -> {
DispatchRate dispatchRate;
switch (type) {
case TOPIC:
dispatchRate = p.topicDispatchRate.get(cluster);
if (dispatchRate == null) {
dispatchRate = p.clusterDispatchRate.get(cluster);
}
break;
case SUBSCRIPTION:
dispatchRate = p.subscriptionDispatchRate.get(cluster);
Expand Down
83 changes: 79 additions & 4 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,14 +18,17 @@
*/
package org.apache.pulsar.client.api;

import com.google.common.collect.Sets;

import static org.testng.Assert.assertNotNull;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
Expand All @@ -37,6 +40,7 @@
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
Expand Down Expand Up @@ -101,6 +105,7 @@ enum DispatchRateType {
*
* @throws Exception
*/
@SuppressWarnings("deprecation")
@Test
public void testMessageRateDynamicallyChange() throws Exception {

Expand All @@ -116,7 +121,7 @@ public void testMessageRateDynamicallyChange() throws Exception {
// (1) verify message-rate is -1 initially
Assert.assertFalse(topic.getDispatchRateLimiter().isPresent());

// (1) change to 100
// (2) change to 100
int messageRate = 100;
DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 360);
admin.namespaces().setDispatchRate(namespace, dispatchRate);
Expand All @@ -134,8 +139,13 @@ public void testMessageRateDynamicallyChange() throws Exception {
}
Assert.assertTrue(isDispatchRateUpdate);
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
Policies policies = admin.namespaces().getPolicies(namespace);
Map<String, DispatchRate> dispatchRateMap = Maps.newHashMap();
dispatchRateMap.put("test", dispatchRate);
Assert.assertEquals(policies.clusterDispatchRate, dispatchRateMap);
Assert.assertEquals(policies.topicDispatchRate, dispatchRateMap);

// (1) change to 500
// (3) change to 500
messageRate = 500;
dispatchRate = new DispatchRate(-1, messageRate, 360);
admin.namespaces().setDispatchRate(namespace, dispatchRate);
Expand All @@ -152,6 +162,10 @@ public void testMessageRateDynamicallyChange() throws Exception {
}
Assert.assertTrue(isDispatchRateUpdate);
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
policies = admin.namespaces().getPolicies(namespace);
dispatchRateMap.put("test", dispatchRate);
Assert.assertEquals(policies.clusterDispatchRate, dispatchRateMap);
Assert.assertEquals(policies.topicDispatchRate, dispatchRateMap);

producer.close();
}
Expand Down Expand Up @@ -896,6 +910,67 @@ public void testClosingRateLimiter(SubscriptionType subscription) throws Excepti
log.info("-- Exiting {} test --", methodName);
}

@SuppressWarnings("deprecation")
@Test
public void testDispatchRateCompatibility1() throws Exception {
final String cluster = "test";

Optional<Policies> policies = Optional.of(new Policies());
DispatchRate clusterDispatchRate = new DispatchRate(100, 512, 1);
DispatchRate topicDispatchRate = new DispatchRate(200, 1024, 1);

// (1) If both clusterDispatchRate and topicDispatchRate are empty, dispatch throttling is disabled
DispatchRate dispatchRate = DispatchRateLimiter.getPoliciesDispatchRate(cluster, policies,
DispatchRateLimiter.Type.TOPIC);
Assert.assertNull(dispatchRate);

// (2) If topicDispatchRate is empty, clusterDispatchRate is effective
policies.get().clusterDispatchRate.put(cluster, clusterDispatchRate);
dispatchRate = DispatchRateLimiter.getPoliciesDispatchRate(cluster, policies, DispatchRateLimiter.Type.TOPIC);
Assert.assertEquals(dispatchRate, clusterDispatchRate);

// (3) If topicDispatchRate is not empty, topicDispatchRate is effective
policies.get().topicDispatchRate.put(cluster, topicDispatchRate);
dispatchRate = DispatchRateLimiter.getPoliciesDispatchRate(cluster, policies, DispatchRateLimiter.Type.TOPIC);
Assert.assertEquals(dispatchRate, topicDispatchRate);
}

@SuppressWarnings("deprecation")
@Test
public void testDispatchRateCompatibility2() throws Exception {
final String namespace = "my-property/dispatch-rate-compatibility";
final String topicName = "persistent://" + namespace + "/t1";
final String cluster = "test";
admin.namespaces().createNamespace(namespace, Sets.newHashSet(cluster));
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
DispatchRateLimiter dispatchRateLimiter = new DispatchRateLimiter(topic, DispatchRateLimiter.Type.TOPIC);

Policies policies = new Policies();
DispatchRate clusterDispatchRate = new DispatchRate(100, 512, 1);
DispatchRate topicDispatchRate = new DispatchRate(200, 1024, 1);

// (1) If both clusterDispatchRate and topicDispatchRate are empty, dispatch throttling is disabled
dispatchRateLimiter.onPoliciesUpdate(policies);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), -1);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), -1);

// (2) If topicDispatchRate is empty, clusterDispatchRate is effective
policies.clusterDispatchRate.put(cluster, clusterDispatchRate);
dispatchRateLimiter.onPoliciesUpdate(policies);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), 100);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), 512);

// (3) If topicDispatchRate is not empty, topicDispatchRate is effective
policies.topicDispatchRate.put(cluster, topicDispatchRate);
dispatchRateLimiter.onPoliciesUpdate(policies);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), 200);
Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), 1024);

producer.close();
topic.close().get();
}

protected void deactiveCursors(ManagedLedgerImpl ledger) throws Exception {
Field statsUpdaterField = BrokerService.class.getDeclaredField("statsUpdater");
statsUpdaterField.setAccessible(true);
Expand Down
6 changes: 5 additions & 1 deletion pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ public class Policies {
public BundlesData bundles;
@SuppressWarnings("checkstyle:MemberName")
public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> backlog_quota_map = Maps.newHashMap();
@Deprecated
public Map<String, DispatchRate> clusterDispatchRate = Maps.newHashMap();
public Map<String, DispatchRate> topicDispatchRate = Maps.newHashMap();
public Map<String, DispatchRate> subscriptionDispatchRate = Maps.newHashMap();
public Map<String, DispatchRate> replicatorDispatchRate = Maps.newHashMap();
Expand Down Expand Up @@ -103,7 +105,7 @@ public class Policies {
@Override
public int hashCode() {
return Objects.hash(auth_policies, replication_clusters,
backlog_quota_map, publishMaxMessageRate,
backlog_quota_map, publishMaxMessageRate, clusterDispatchRate,
topicDispatchRate, subscriptionDispatchRate, replicatorDispatchRate,
clusterSubscribeRate, deduplicationEnabled, persistence,
bundles, latency_stats_sample_rate,
Expand All @@ -128,6 +130,7 @@ public boolean equals(Object obj) {
return Objects.equals(auth_policies, other.auth_policies)
&& Objects.equals(replication_clusters, other.replication_clusters)
&& Objects.equals(backlog_quota_map, other.backlog_quota_map)
&& Objects.equals(clusterDispatchRate, other.clusterDispatchRate)
&& Objects.equals(topicDispatchRate, other.topicDispatchRate)
&& Objects.equals(subscriptionDispatchRate, other.subscriptionDispatchRate)
&& Objects.equals(replicatorDispatchRate, other.replicatorDispatchRate)
Expand Down Expand Up @@ -182,6 +185,7 @@ public String toString() {
.add("replication_clusters", replication_clusters).add("bundles", bundles)
.add("backlog_quota_map", backlog_quota_map).add("persistence", persistence)
.add("deduplicationEnabled", deduplicationEnabled)
.add("clusterDispatchRate", clusterDispatchRate)
.add("topicDispatchRate", topicDispatchRate)
.add("subscriptionDispatchRate", subscriptionDispatchRate)
.add("replicatorDispatchRate", replicatorDispatchRate)
Expand Down