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 2018/04/02 17:40:15 UTC
[GitHub] merlimat closed pull request #1358: Issue 1071: add ratelimiter for subscription
merlimat closed pull request #1358: Issue 1071: add ratelimiter for subscription
URL: https://github.com/apache/incubator-pulsar/pull/1358
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 0ad3fb977..2e09ba9ad 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -146,6 +146,14 @@
// default message-byte dispatch-throttling
@FieldContext(dynamic = true)
private long dispatchThrottlingRatePerTopicInByte = 0;
+ // Default number of message dispatching throttling-limit for a subscription.
+ // Using a value of 0, is disabling.
+ @FieldContext(dynamic = true)
+ private int dispatchThrottlingRatePerSubscriptionInMsg = 0;
+ // Default number of message-bytes dispatching throttling-limit for a subscription.
+ // Using a value of 0, is disabling.
+ @FieldContext(dynamic = true)
+ private long dispatchThrottlingRatePerSubscribeInByte = 0;
// Default dispatch-throttling is disabled for consumers which already caught-up with published messages and
// don't have backlog. This enables dispatch-throttling for non-backlog consumers as well.
@FieldContext(dynamic = true)
@@ -705,6 +713,22 @@ public void setDispatchThrottlingRatePerTopicInByte(long dispatchThrottlingRateP
this.dispatchThrottlingRatePerTopicInByte = dispatchThrottlingRatePerTopicInByte;
}
+ public int getDispatchThrottlingRatePerSubscriptionInMsg() {
+ return dispatchThrottlingRatePerSubscriptionInMsg;
+ }
+
+ public void setDispatchThrottlingRatePerSubscriptionInMsg(int dispatchThrottlingRatePerSubscriptionInMsg) {
+ this.dispatchThrottlingRatePerSubscriptionInMsg = dispatchThrottlingRatePerSubscriptionInMsg;
+ }
+
+ public long getDispatchThrottlingRatePerSubscribeInByte() {
+ return dispatchThrottlingRatePerSubscribeInByte;
+ }
+
+ public void setDispatchThrottlingRatePerSubscribeInByte(long dispatchThrottlingRatePerSubscribeInByte) {
+ this.dispatchThrottlingRatePerSubscribeInByte = dispatchThrottlingRatePerSubscribeInByte;
+ }
+
public boolean isDispatchThrottlingOnNonBacklogConsumerEnabled() {
return dispatchThrottlingOnNonBacklogConsumerEnabled;
}
@@ -1503,7 +1527,7 @@ public void setTlsProtocols(Set<String> tlsProtocols) {
public void setTlsCiphers(Set<String> tlsCiphers) {
this.tlsCiphers = tlsCiphers;
}
-
+
public boolean getTlsRequireTrustedClientCertOnConnect() {
return tlsRequireTrustedClientCertOnConnect;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index c71127996..2f2f63ea3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -633,6 +633,56 @@ protected DispatchRate internalGetDispatchRate() {
}
}
+ protected void internalSetSubscriptionDispatchRate(DispatchRate dispatchRate) {
+ log.info("[{}] Set namespace subscription dispatch-rate {}/{}", clientAppId(), namespaceName, dispatchRate);
+ validateSuperUserAccess();
+
+ Entry<Policies, Stat> policiesNode = null;
+
+ try {
+ final String path = path(POLICIES, namespaceName.toString());
+ // Force to read the data s.t. the watch to the cache content is setup.
+ policiesNode = policiesCache().getWithStat(path).orElseThrow(
+ () -> new RestException(Status.NOT_FOUND, "Namespace " + namespaceName + " does not exist"));
+ policiesNode.getKey().subscriptionDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);
+
+ // Write back the new policies into zookeeper
+ globalZk().setData(path, jsonMapper().writeValueAsBytes(policiesNode.getKey()),
+ policiesNode.getValue().getVersion());
+ policiesCache().invalidate(path);
+
+ log.info("[{}] Successfully updated the subscriptionDispatchRate for cluster on namespace {}", clientAppId(),
+ namespaceName);
+ } catch (KeeperException.NoNodeException e) {
+ log.warn("[{}] Failed to update the subscriptionDispatchRate for cluster on namespace {}: does not exist",
+ clientAppId(), namespaceName);
+ throw new RestException(Status.NOT_FOUND, "Namespace does not exist");
+ } catch (KeeperException.BadVersionException e) {
+ log.warn(
+ "[{}] Failed to update the subscriptionDispatchRate for cluster on namespace {} expected policy node version={} : concurrent modification",
+ clientAppId(), namespaceName, policiesNode.getValue().getVersion());
+
+ throw new RestException(Status.CONFLICT, "Concurrent modification");
+ } catch (Exception e) {
+ log.error("[{}] Failed to update the subscriptionDispatchRate for cluster on namespace {}", clientAppId(),
+ namespaceName, e);
+ throw new RestException(e);
+ }
+ }
+
+ protected DispatchRate internalGetSubscriptionDispatchRate() {
+ validateAdminAccessOnProperty(namespaceName.getProperty());
+
+ Policies policies = getNamespacePolicies(namespaceName);
+ DispatchRate dispatchRate = policies.subscriptionDispatchRate.get(pulsar().getConfiguration().getClusterName());
+ if (dispatchRate != null) {
+ return dispatchRate;
+ } else {
+ throw new RestException(Status.NOT_FOUND,
+ "Subscription-Dispatch-rate is not configured for cluster " + pulsar().getConfiguration().getClusterName());
+ }
+ }
+
protected void internalSetBacklogQuota(BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota) {
validateAdminAccessOnProperty(namespaceName.getProperty());
validatePoliciesReadOnlyAccess();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
index e3f5e4338..eae555951 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
@@ -518,6 +518,30 @@ public DispatchRate getDispatchRate(@PathParam("property") String property, @Pat
return internalGetDispatchRate();
}
+ @POST
+ @Path("/{property}/{cluster}/{namespace}/subscriptionDispatchRate")
+ @ApiOperation(value = "Set Subscription dispatch-rate throttling for all topics of the namespace")
+ @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+ public void setSubscriptionDispatchRate(@PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace,
+ DispatchRate dispatchRate) {
+ validateNamespaceName(property, cluster, namespace);
+ internalSetSubscriptionDispatchRate(dispatchRate);
+ }
+
+ @GET
+ @Path("/{property}/{cluster}/{namespace}/subscriptionDispatchRate")
+ @ApiOperation(value = "Get Subscription dispatch-rate configured for the namespace, -1 represents not configured yet")
+ @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Namespace does not exist") })
+ public DispatchRate getSubscriptionDispatchRate(@PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace) {
+ validateNamespaceName(property, cluster, namespace);
+ return internalGetSubscriptionDispatchRate();
+ }
+
@GET
@Path("/{property}/{cluster}/{namespace}/backlogQuotaMap")
@ApiOperation(hidden = true, value = "Get backlog quota map on a namespace.")
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index c984e7681..a22baaced 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -319,6 +319,28 @@ public DispatchRate getDispatchRate(@PathParam("property") String property,
return internalGetDispatchRate();
}
+ @POST
+ @Path("/{property}/{namespace}/subscriptionDispatchRate")
+ @ApiOperation(value = "Set Subscription dispatch-rate throttling for all topics of the namespace")
+ @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+ public void setSubscriptionDispatchRate(@PathParam("property") String property,
+ @PathParam("namespace") String namespace,
+ DispatchRate dispatchRate) {
+ validateNamespaceName(property, namespace);
+ internalSetSubscriptionDispatchRate(dispatchRate);
+ }
+
+ @GET
+ @Path("/{property}/{namespace}/subscriptionDispatchRate")
+ @ApiOperation(value = "Get Subscription dispatch-rate configured for the namespace, -1 represents not configured yet")
+ @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Namespace does not exist") })
+ public DispatchRate getSubscriptionDispatchRate(@PathParam("property") String property,
+ @PathParam("namespace") String namespace) {
+ validateNamespaceName(property, namespace);
+ return internalGetSubscriptionDispatchRate();
+ }
+
@GET
@Path("/{property}/{namespace}/backlogQuotaMap")
@ApiOperation(value = "Get backlog quota map on a namespace.")
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index f77e41875..e1a0d3f0f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -72,7 +72,9 @@
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
import org.apache.pulsar.broker.web.PulsarWebResource;
@@ -1090,6 +1092,14 @@ private void updateConfigurationAndRegisterListeners() {
registerConfigurationListener("autoSkipNonRecoverableData", (skipNonRecoverableLedger) -> {
updateManagedLedgerConfig();
});
+ // add listener to update message-dispatch-rate in msg
+ registerConfigurationListener("dispatchThrottlingRatePerSubscriptionInMsg", (dispatchRatePerTopicInMsg) -> {
+ updateSubscriptionMessageDispatchRate();
+ });
+ // add listener to update message-dispatch-rate in byte
+ registerConfigurationListener("dispatchThrottlingRatePerSubscribeInByte", (dispatchRatePerTopicInByte) -> {
+ updateSubscriptionMessageDispatchRate();
+ });
// add more listeners here
}
@@ -1114,6 +1124,31 @@ private void updateTopicMessageDispatchRate() {
});
}
+ private void updateSubscriptionMessageDispatchRate() {
+ this.pulsar().getExecutor().submit(() -> {
+ // update message-rate for each topic subscription
+ topics.forEach((name, topicFuture) -> {
+ if (topicFuture.isDone()) {
+ try {
+ topicFuture.get().getSubscriptions().forEach((subName, persistentSubscription) -> {
+ if (persistentSubscription
+ .getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
+ ((PersistentDispatcherMultipleConsumers) persistentSubscription
+ .getDispatcher()).getDispatchRateLimiter().updateDispatchRate();
+ } else if (persistentSubscription
+ .getDispatcher() instanceof PersistentDispatcherSingleActiveConsumer) {
+ ((PersistentDispatcherSingleActiveConsumer) persistentSubscription
+ .getDispatcher()).getDispatchRateLimiter().updateDispatchRate();
+ }
+ });
+ } catch (Exception e) {
+ log.warn("Failed to get topic from future while update subscription dispatch rate ", e);
+ }
+ }
+ });
+ });
+ }
+
private void updateManagedLedgerConfig() {
this.pulsar().getExecutor().submit(() -> {
// update managed-ledger config of each topic
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
index 55dda8326..e511d40f3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
@@ -38,20 +38,26 @@
public class DispatchRateLimiter {
private final String topicName;
+ private final String subscriptionName;
private final BrokerService brokerService;
private RateLimiter dispatchRateLimiterOnMessage;
private RateLimiter dispatchRateLimiterOnByte;
- public DispatchRateLimiter(PersistentTopic topic) {
+ public DispatchRateLimiter(PersistentTopic topic, String subscriptionName) {
this.topicName = topic.getName();
+ this.subscriptionName = subscriptionName;
this.brokerService = topic.getBrokerService();
updateDispatchRate();
registerLocalPoliciesListener();
}
+ public DispatchRateLimiter(PersistentTopic topic) {
+ this(topic, null);
+ }
+
/**
* returns available msg-permit if msg-dispatch-throttling is enabled else it returns -1
- *
+ *
* @return
*/
public long getAvailableDispatchRateLimitOnMsg() {
@@ -60,7 +66,7 @@ public long getAvailableDispatchRateLimitOnMsg() {
/**
* It acquires msg and bytes permits from rate-limiter and returns if acquired permits succeed.
- *
+ *
* @param msgPermits
* @param bytePermits
* @return
@@ -77,7 +83,7 @@ public boolean tryDispatchPermit(long msgPermits, long bytePermits) {
/**
* checks if dispatch-rate limit is configured and if it's configured then check if permits are available or not.
- *
+ *
* @return
*/
public boolean hasMessageDispatchPermit() {
@@ -87,7 +93,7 @@ public boolean hasMessageDispatchPermit() {
/**
* Checks if dispatch-rate limiting is enabled.
- *
+ *
* @return
*/
public boolean isDispatchRateLimitingEnabled() {
@@ -101,16 +107,21 @@ public boolean isDispatchRateLimitingEnabled() {
public void updateDispatchRate() {
DispatchRate dispatchRate = getPoliciesDispatchRate();
if (dispatchRate == null) {
- dispatchRate = new DispatchRate(brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(),
+ if (subscriptionName == null) {
+ dispatchRate = new DispatchRate(brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(),
brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInByte(), 1);
+ } else {
+ dispatchRate = new DispatchRate(brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg(),
+ brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerSubscribeInByte(), 1);
+ }
}
updateDispatchRate(dispatchRate);
- log.info("[{}] configured message-dispatch rate at broker {}", this.topicName, dispatchRate);
+ log.info("[{}] [{}] configured message-dispatch rate at broker {}", this.topicName, this.subscriptionName, dispatchRate);
}
/**
* Register listener on namespace policy change to update dispatch-rate if required
- *
+ *
*/
private void registerLocalPoliciesListener() {
brokerService.pulsar().getConfigurationCache().policiesCache().registerListener((path, data, stat) -> {
@@ -118,16 +129,25 @@ private void registerLocalPoliciesListener() {
final String cluster = brokerService.pulsar().getConfiguration().getClusterName();
final String policiesPath = path(POLICIES, namespace.toString());
if (policiesPath.equals(path)) {
- DispatchRate dispatchRate = data.clusterDispatchRate.get(cluster);
+ DispatchRate dispatchRate;
+ if (subscriptionName == null) {
+ dispatchRate = data.clusterDispatchRate.get(cluster);
+ } else {
+ dispatchRate = data.subscriptionDispatchRate.get(cluster);
+ }
// update dispatch-rate only if it's configured in policies else ignore
if (dispatchRate != null) {
- final DispatchRate clusterDispatchRate = new DispatchRate(
- brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(),
- brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInByte(), 1);
+ int inMsg = (subscriptionName == null) ?
+ brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg() :
+ brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg();
+ long inByte = (subscriptionName == null) ?
+ brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInByte() :
+ brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerSubscribeInByte();
+ final DispatchRate newDispatchRate = new DispatchRate(inMsg, inByte, 1);
// if policy-throttling rate is disabled and cluster-throttling is enabled then apply
// cluster-throttling rate
- if (!isDispatchRateEnabled(dispatchRate) && isDispatchRateEnabled(clusterDispatchRate)) {
- dispatchRate = clusterDispatchRate;
+ if (!isDispatchRateEnabled(dispatchRate) && isDispatchRateEnabled(newDispatchRate)) {
+ dispatchRate = newDispatchRate;
}
updateDispatchRate(dispatchRate);
}
@@ -137,7 +157,7 @@ private void registerLocalPoliciesListener() {
/**
* Gets configured dispatch-rate from namespace policies. Returns null if dispatch-rate is not configured
- *
+ *
* @return
*/
public DispatchRate getPoliciesDispatchRate() {
@@ -149,11 +169,16 @@ public DispatchRate getPoliciesDispatchRate() {
policies = brokerService.pulsar().getConfigurationCache().policiesCache().getAsync(path)
.get(cacheTimeOutInSec, SECONDS);
} catch (Exception e) {
- log.warn("Failed to get message-rate for {}", this.topicName, e);
+ log.warn("Failed to get message-rate for {} subscription {}", this.topicName, this.subscriptionName, e);
}
// return policy-dispatch rate only if it's enabled in policies
return policies.map(p -> {
- DispatchRate dispatchRate = p.clusterDispatchRate.get(cluster);
+ DispatchRate dispatchRate;
+ if (subscriptionName == null) {
+ dispatchRate = p.clusterDispatchRate.get(cluster);
+ } else {
+ dispatchRate = p.subscriptionDispatchRate.get(cluster);
+ }
return isDispatchRateEnabled(dispatchRate) ? dispatchRate : null;
}).orElse(null);
}
@@ -161,22 +186,22 @@ public DispatchRate getPoliciesDispatchRate() {
/**
* Update dispatch rate by updating msg and byte rate-limiter. If dispatch-rate is configured < 0 then it closes
* the rate-limiter and disables appropriate rate-limiter.
- *
+ *
* @param dispatchRate
*/
public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
// synchronized to prevent race condition from concurrent zk-watch
- log.info("[{}] setting message-dispatch-rate {}", topicName, dispatchRate);
+ log.info("[{}] [{}] setting message-dispatch-rate {}", topicName, subscriptionName, dispatchRate);
- long msgRate = dispatchRate.dispatchThrottlingRatePerTopicInMsg;
- long byteRate = dispatchRate.dispatchThrottlingRatePerTopicInByte;
- long ratePerid = dispatchRate.ratePeriodInSecond;
+ long msgRate = dispatchRate.dispatchThrottlingRateInMsg;
+ long byteRate = dispatchRate.dispatchThrottlingRateInByte;
+ long ratePeriod = dispatchRate.ratePeriodInSecond;
// update msg-rateLimiter
if (msgRate > 0) {
if (this.dispatchRateLimiterOnMessage == null) {
this.dispatchRateLimiterOnMessage = new RateLimiter(brokerService.pulsar().getExecutor(), msgRate,
- ratePerid, TimeUnit.SECONDS);
+ ratePeriod, TimeUnit.SECONDS);
} else {
this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.ratePeriodInSecond,
TimeUnit.SECONDS);
@@ -193,7 +218,7 @@ public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
if (byteRate > 0) {
if (this.dispatchRateLimiterOnByte == null) {
this.dispatchRateLimiterOnByte = new RateLimiter(brokerService.pulsar().getExecutor(), byteRate,
- ratePerid, TimeUnit.SECONDS);
+ ratePeriod, TimeUnit.SECONDS);
} else {
this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.ratePeriodInSecond,
TimeUnit.SECONDS);
@@ -209,7 +234,7 @@ public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
/**
* Get configured msg dispatch-throttling rate. Returns -1 if not configured
- *
+ *
* @return
*/
public long getDispatchRateOnMsg() {
@@ -218,7 +243,7 @@ public long getDispatchRateOnMsg() {
/**
* Get configured byte dispatch-throttling rate. Returns -1 if not configured
- *
+ *
* @return
*/
public long getDispatchRateOnByte() {
@@ -227,8 +252,8 @@ public long getDispatchRateOnByte() {
private boolean isDispatchRateEnabled(DispatchRate dispatchRate) {
- return dispatchRate != null && (dispatchRate.dispatchThrottlingRatePerTopicInMsg > 0
- || dispatchRate.dispatchThrottlingRatePerTopicInByte > 0);
+ return dispatchRate != null && (dispatchRate.dispatchThrottlingRateInMsg > 0
+ || dispatchRate.dispatchThrottlingRateInByte > 0);
}
public void close() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 54e40b6f3..2eb83ae29 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -86,6 +86,7 @@
private static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers> BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class, "blockedDispatcherOnUnackedMsgs");
private final ServiceConfiguration serviceConfig;
+ private DispatchRateLimiter dispatchRateLimiter;
enum ReadType {
Normal, Replay
@@ -100,6 +101,7 @@ public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCurso
this.maxUnackedMessages = topic.getBrokerService().pulsar().getConfiguration()
.getMaxUnackedMessagesPerSubscription();
this.serviceConfig = topic.getBrokerService().pulsar().getConfiguration();
+ this.dispatchRateLimiter = null;
}
@Override
@@ -231,12 +233,12 @@ public void readMoreEntries() {
// active-cursor reads message from cache rather from bookkeeper (2) if topic has reached message-rate
// threshold: then schedule the read after MESSAGE_RATE_BACKOFF_MS
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
- DispatchRateLimiter rateLimiter = topic.getDispatchRateLimiter();
- if (rateLimiter.isDispatchRateLimitingEnabled()) {
- if (!rateLimiter.hasMessageDispatchPermit()) {
+ DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter();
+ if (topicRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!topicRateLimiter.hasMessageDispatchPermit()) {
if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded message-rate {}/{}, schedule after a {}", name,
- rateLimiter.getDispatchRateOnMsg(), rateLimiter.getDispatchRateOnByte(),
+ log.debug("[{}] message-read exceeded topic message-rate {}/{}, schedule after a {}", name,
+ topicRateLimiter.getDispatchRateOnMsg(), topicRateLimiter.getDispatchRateOnByte(),
MESSAGE_RATE_BACKOFF_MS);
}
topic.getBrokerService().executor().schedule(() -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS,
@@ -244,12 +246,35 @@ public void readMoreEntries() {
return;
} else {
// if dispatch-rate is in msg then read only msg according to available permit
- long availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
+ long availablePermitsOnMsg = topicRateLimiter.getAvailableDispatchRateLimitOnMsg();
if (availablePermitsOnMsg > 0) {
messagesToRead = Math.min(messagesToRead, (int) availablePermitsOnMsg);
}
}
}
+
+ if (dispatchRateLimiter == null) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ if (dispatchRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!dispatchRateLimiter.hasMessageDispatchPermit()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] message-read exceeded subscription message-rate {}/{}, schedule after a {}", name,
+ dispatchRateLimiter.getDispatchRateOnMsg(), dispatchRateLimiter.getDispatchRateOnByte(),
+ MESSAGE_RATE_BACKOFF_MS);
+ }
+ topic.getBrokerService().executor().schedule(() -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS,
+ TimeUnit.MILLISECONDS);
+ return;
+ } else {
+ // if dispatch-rate is in msg then read only msg according to available permit
+ long availablePermitsOnMsg = dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg();
+ if (availablePermitsOnMsg > 0) {
+ messagesToRead = Math.min(messagesToRead, (int) availablePermitsOnMsg);
+ }
+ }
+ }
+
}
if (!messagesToReplay.isEmpty()) {
@@ -418,6 +443,11 @@ public synchronized void readEntriesComplete(List<Entry> entries, Object ctx) {
// acquire message-dispatch permits for already delivered messages
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
topic.getDispatchRateLimiter().tryDispatchPermit(totalMessagesSent, totalBytesSent);
+
+ if (dispatchRateLimiter == null) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ dispatchRateLimiter.tryDispatchPermit(totalMessagesSent, totalBytesSent);
}
if (entriesToDispatch > 0) {
@@ -586,5 +616,13 @@ public String getName() {
return name;
}
+ public DispatchRateLimiter getDispatchRateLimiter() {
+ if ((serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) &&
+ (dispatchRateLimiter == null)) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ return dispatchRateLimiter;
+ }
+
private static final Logger log = LoggerFactory.getLogger(PersistentDispatcherMultipleConsumers.class);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index cc30836d4..29426ffdd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -52,6 +52,7 @@
private final PersistentTopic topic;
private final ManagedCursor cursor;
private final String name;
+ private DispatchRateLimiter dispatchRateLimiter;
private boolean havePendingRead = false;
@@ -70,6 +71,7 @@ public PersistentDispatcherSingleActiveConsumer(ManagedCursor cursor, SubType su
this.cursor = cursor;
this.readBatchSize = MaxReadBatchSize;
this.serviceConfig = topic.getBrokerService().pulsar().getConfiguration();
+ this.dispatchRateLimiter = null;
}
protected void scheduleReadOnActiveConsumer() {
@@ -201,6 +203,11 @@ public synchronized void readEntriesComplete(final List<Entry> entries, Object o
// acquire message-dispatch permits for already delivered messages
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
topic.getDispatchRateLimiter().tryDispatchPermit(totalMessagesSent, totalBytesSent);
+
+ if (dispatchRateLimiter == null) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ dispatchRateLimiter.tryDispatchPermit(totalMessagesSent, totalBytesSent);
}
// Schedule a new read batch operation only after the previous batch has been written to the socket
synchronized (PersistentDispatcherSingleActiveConsumer.this) {
@@ -307,13 +314,13 @@ protected void readMoreEntries(Consumer consumer) {
// active-cursor reads message from cache rather from bookkeeper (2) if topic has reached message-rate
// threshold: then schedule the read after MESSAGE_RATE_BACKOFF_MS
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
- DispatchRateLimiter rateLimiter = topic.getDispatchRateLimiter();
- if (rateLimiter.isDispatchRateLimitingEnabled()) {
- if (!rateLimiter.hasMessageDispatchPermit()) {
+ DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter();
+ if (topicRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!topicRateLimiter.hasMessageDispatchPermit()) {
if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded message-rate {}/{}, schedule after a {}", name,
- rateLimiter.getDispatchRateOnMsg(), rateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
+ log.debug("[{}] message-read exceeded topic message-rate {}/{}, schedule after a {}", name,
+ topicRateLimiter.getDispatchRateOnMsg(), topicRateLimiter.getDispatchRateOnByte(),
+ MESSAGE_RATE_BACKOFF_MS);
}
topic.getBrokerService().executor().schedule(() -> {
Consumer currentConsumer = ACTIVE_CONSUMER_UPDATER.get(this);
@@ -321,19 +328,49 @@ protected void readMoreEntries(Consumer consumer) {
readMoreEntries(currentConsumer);
} else {
if (log.isDebugEnabled()) {
- log.info("[{}] Skipping read retry: Current Consumer {}, havePendingRead {}",
- topic.getName(), currentConsumer, havePendingRead);
+ log.debug("[{}] Skipping read retry for topic: Current Consumer {}, havePendingRead {}",
+ topic.getName(), currentConsumer, havePendingRead);
}
}
}, MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
return;
} else {
// if dispatch-rate is in msg then read only msg according to available permit
- long availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
+ long availablePermitsOnMsg = topicRateLimiter.getAvailableDispatchRateLimitOnMsg();
if (availablePermitsOnMsg > 0) {
messagesToRead = Math.min(messagesToRead, (int) availablePermitsOnMsg);
}
+ }
+ }
+ if (dispatchRateLimiter == null) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ if (dispatchRateLimiter.isDispatchRateLimitingEnabled()) {
+ if (!dispatchRateLimiter.hasMessageDispatchPermit()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] message-read exceeded subscription message-rate {}/{}, schedule after a {}", name,
+ dispatchRateLimiter.getDispatchRateOnMsg(), dispatchRateLimiter.getDispatchRateOnByte(),
+ MESSAGE_RATE_BACKOFF_MS);
+ }
+ topic.getBrokerService().executor().schedule(() -> {
+ Consumer currentConsumer = ACTIVE_CONSUMER_UPDATER.get(this);
+ if (currentConsumer != null && !havePendingRead) {
+ readMoreEntries(currentConsumer);
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Skipping read retry: Current Consumer {}, havePendingRead {}",
+ topic.getName(), currentConsumer, havePendingRead);
+ }
+ }
+ }, MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
+ return;
+ } else {
+ // if dispatch-rate is in msg then read only msg according to available permit
+ long subPermitsOnMsg = dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg();
+ if (subPermitsOnMsg > 0) {
+ messagesToRead = Math.min(messagesToRead, (int) subPermitsOnMsg);
+ }
}
}
}
@@ -402,6 +439,14 @@ public synchronized void readEntriesFailed(ManagedLedgerException exception, Obj
}
+ public DispatchRateLimiter getDispatchRateLimiter() {
+ if ((serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) &&
+ (dispatchRateLimiter == null)) {
+ dispatchRateLimiter = new DispatchRateLimiter(topic, name);
+ }
+ return dispatchRateLimiter;
+ }
+
@Override
public void addUnAckedMessages(int unAckMessages) {
// No-op
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
index 44257660c..59e343933 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
@@ -214,7 +214,7 @@ public void testMessageRateLimitingNotReceiveAllMessages(SubscriptionType subscr
producer.send(new byte[80]);
}
- // consumer should not have received all publihsed message due to message-rate throttling
+ // consumer should not have received all published message due to message-rate throttling
Assert.assertTrue(totalReceived.get() < messageRate * 2);
consumer.close();
@@ -763,7 +763,7 @@ public void testNonBacklogConsumerWithThrottlingEnabled(SubscriptionType subscri
producer.send(new byte[80]);
}
- // consumer should not have received all publihsed message due to message-rate throttling
+ // consumer should not have received all published message due to message-rate throttling
Assert.assertTrue(totalReceived.get() < messageRate * 2);
consumer.close();
@@ -847,7 +847,7 @@ public void testClusterPolicyOverrideConfiguration() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- private void deactiveCursors(ManagedLedgerImpl ledger) throws Exception {
+ protected void deactiveCursors(ManagedLedgerImpl ledger) throws Exception {
Field statsUpdaterField = BrokerService.class.getDeclaredField("statsUpdater");
statsUpdaterField.setAccessible(true);
ScheduledExecutorService statsUpdater = (ScheduledExecutorService) statsUpdaterField
@@ -858,4 +858,4 @@ private void deactiveCursors(ManagedLedgerImpl ledger) throws Exception {
});
}
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
new file mode 100644
index 000000000..2ed6e11f9
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
@@ -0,0 +1,542 @@
+/**
+ * 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.client.api;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.common.policies.data.DispatchRate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchThrottlingTest {
+ private static final Logger log = LoggerFactory.getLogger(SubscriptionMessageDispatchThrottlingTest.class);
+
+ /**
+ * verify: consumer should not receive all messages due to message-rate throttling
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptionAndDispatchRateType", timeOut = 5000)
+ public void testMessageRateLimitingNotReceiveAllMessages(SubscriptionType subscription,
+ DispatchRateType dispatchRateType) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName = "persistent://" + namespace + "/throttlingBlock";
+ final String subName = "my-subscriber-name";
+
+ final int messageRate = 100;
+ DispatchRate dispatchRate = null;
+ if (DispatchRateType.messageRate.equals(dispatchRateType)) {
+ dispatchRate = new DispatchRate(messageRate, -1, 360);
+ } else {
+ dispatchRate = new DispatchRate(-1, messageRate, 360);
+ }
+
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+ // create producer, topic and consumer
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get();
+
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ }).subscribe();
+
+ DispatchRateLimiter subRateLimiter = null;
+ Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ boolean isMessageRateUpdate = false;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() > 0
+ || subRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+ Assert.assertEquals(admin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
+
+ int numMessages = 500;
+ // Asynchronously produce messages
+ for (int i = 0; i < numMessages; i++) {
+ producer.send(new byte[80]);
+ }
+
+ // consumer should not have received all publihsed message due to message-rate throttling
+ Assert.assertTrue(totalReceived.get() < messageRate * 2);
+
+ consumer.close();
+ producer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ /**
+ * verify rate-limiting should throttle message-dispatching based on message-rate
+ *
+ * <pre>
+ * 1. dispatch-msg-rate = 10 msg/sec
+ * 2. send 30 msgs
+ * 3. it should take up to 2 second to receive all messages
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 5000)
+ public void testMessageRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription)
+ throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName = "persistent://" + namespace + "/throttlingAll";
+ final String subName = "my-subscriber-name";
+
+ final int messageRate = 10;
+ DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 1);
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+ final int numProducedMessages = 30;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get();
+
+ DispatchRateLimiter subRateLimiter = null;
+ Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ boolean isMessageRateUpdate = false;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() > 0
+ || subRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+ Assert.assertEquals(admin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessages; i++) {
+ final String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+ latch.await();
+ Assert.assertEquals(totalReceived.get(), numProducedMessages);
+ long end = System.currentTimeMillis();
+
+ Assert.assertTrue((end - start) >= 2000);
+
+ consumer.close();
+ producer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ /**
+ * verify rate-limiting should throttle message-dispatching based on byte-rate
+ *
+ * <pre>
+ * 1. dispatch-byte-rate = 100 bytes/sec
+ * 2. send 30 msgs : each with 10 byte
+ * 3. it should take up to 2 second to receive all messages
+ * </pre>
+ *
+ * @param subscription
+ * @throws Exception
+ */
+ @Test(dataProvider = "subscriptions", timeOut = 5000)
+ public void testBytesRateLimitingReceiveAllMessagesAfterThrottling(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName = "persistent://" + namespace + "/throttlingAll";
+ final String subName = "my-subscriber-name";
+
+ final int byteRate = 100;
+ DispatchRate dispatchRate = new DispatchRate(-1, byteRate, 1);
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+ final int numProducedMessages = 30;
+ final CountDownLatch latch = new CountDownLatch(numProducedMessages);
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).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 the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ latch.countDown();
+ }).subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get();
+
+ DispatchRateLimiter subRateLimiter = null;
+ Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ boolean isMessageRateUpdate = false;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() > 0
+ || subRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+ Assert.assertEquals(admin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
+
+ long start = System.currentTimeMillis();
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessages; i++) {
+ producer.send(new byte[byteRate / 10]);
+ }
+ latch.await();
+ Assert.assertEquals(totalReceived.get(), numProducedMessages);
+ long end = System.currentTimeMillis();
+ log.info("-- end - start: {} ", end - start);
+
+ // first 10 messages, which equals receiverQueueSize, will not wait.
+ Assert.assertTrue((end - start) >= 2000);
+
+ consumer.close();
+ producer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ /**
+ * verify message-rate on multiple consumers with shared-subscription
+ *
+ * @throws Exception
+ */
+ @Test(timeOut = 5000)
+ public void testRateLimitingMultipleConsumers() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName = "persistent://" + namespace + "/throttlingMultipleConsumers";
+ final String subName = "my-subscriber-name";
+
+ final int messageRate = 5;
+ DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 360);
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+
+ final int numProducedMessages = 500;
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ ConsumerBuilder<byte[]> consumerBuilder = pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName(subName).subscriptionType(SubscriptionType.Shared).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ });
+ Consumer<byte[]> consumer1 = consumerBuilder.subscribe();
+ Consumer<byte[]> consumer2 = consumerBuilder.subscribe();
+ Consumer<byte[]> consumer3 = consumerBuilder.subscribe();
+ Consumer<byte[]> consumer4 = consumerBuilder.subscribe();
+ Consumer<byte[]> consumer5 = consumerBuilder.subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get();
+
+ DispatchRateLimiter subRateLimiter = null;
+ Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ boolean isMessageRateUpdate = false;
+ int retry = 5;
+ for (int i = 0; i < retry; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() > 0
+ || subRateLimiter.getDispatchRateOnByte() > 0) {
+ isMessageRateUpdate = true;
+ break;
+ } else {
+ if (i != retry - 1) {
+ Thread.sleep(100);
+ }
+ }
+ }
+ Assert.assertTrue(isMessageRateUpdate);
+ Assert.assertEquals(admin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
+
+ // Asynchronously produce messages
+ for (int i = 0; i < numProducedMessages; i++) {
+ final String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+
+ // it can make sure that consumer had enough time to consume message but couldn't consume due to throttling
+ Thread.sleep(500);
+
+ // consumer should not have received all published message due to message-rate throttling
+ Assert.assertNotEquals(totalReceived.get(), numProducedMessages);
+
+ consumer1.close();
+ consumer2.close();
+ consumer3.close();
+ consumer4.close();
+ consumer5.close();
+ producer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+
+ @Test(dataProvider = "subscriptions", timeOut = 5000)
+ public void testClusterRateLimitingConfiguration(SubscriptionType subscription) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName = "persistent://" + namespace + "/throttlingBlock";
+ final String subName = "my-subscriber-name";
+
+ final int messageRate = 5;
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ int initValue = pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg();
+ // (1) Update message-dispatch-rate limit
+ admin.brokers().updateDynamicConfiguration("dispatchThrottlingRatePerSubscriptionInMsg",
+ Integer.toString(messageRate));
+ // sleep incrementally as zk-watch notification is async and may take some time
+ for (int i = 0; i < 5; i++) {
+ if (pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg() == initValue) {
+ Thread.sleep(50 + (i * 10));
+ }
+ }
+ Assert.assertNotEquals(pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg(), initValue);
+
+ admin.namespaces().createNamespace(namespace);
+ // create producer and topic
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get();
+ int numMessages = 500;
+
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
+ .subscriptionType(subscription).messageListener((c1, msg) -> {
+ Assert.assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message [{}] in the listener", receivedMessage);
+ totalReceived.incrementAndGet();
+ }).subscribe();
+
+ // Asynchronously produce messages
+ for (int i = 0; i < numMessages; i++) {
+ final String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+
+ // it can make sure that consumer had enough time to consume message but couldn't consume due to throttling
+ Thread.sleep(500);
+
+ // consumer should not have received all published message due to message-rate throttling
+ Assert.assertNotEquals(totalReceived.get(), numMessages);
+
+ consumer.close();
+ producer.close();
+ pulsar.getConfiguration().setDispatchThrottlingRatePerSubscriptionInMsg(initValue);
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ /**
+ * <pre>
+ * It verifies that cluster-throttling value gets considered when namespace-policy throttling is disabled.
+ *
+ * 1. Update cluster-throttling-config: topic rate-limiter has cluster-config
+ * 2. Update namespace-throttling-config: topic rate-limiter has namespace-config
+ * 3. Disable namespace-throttling-config: topic rate-limiter has cluster-config
+ * 4. Create new topic with disable namespace-config and enabled cluster-config: it takes cluster-config
+ *
+ * </pre>
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testClusterPolicyOverrideConfiguration() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final String namespace = "my-property/use/throttling_ns";
+ final String topicName1 = "persistent://" + namespace + "/throttlingOverride1";
+ final String topicName2 = "persistent://" + namespace + "/throttlingOverride2";
+ final String subName1 = "my-subscriber-name1";
+ final String subName2 = "my-subscriber-name2";
+
+ final int clusterMessageRate = 100;
+ // enable throttling for nonBacklog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+
+ int initValue = pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg();
+ // (1) Update message-dispatch-rate limit
+ admin.brokers().updateDynamicConfiguration("dispatchThrottlingRatePerSubscriptionInMsg",
+ Integer.toString(clusterMessageRate));
+ // sleep incrementally as zk-watch notification is async and may take some time
+ for (int i = 0; i < 5; i++) {
+ if (pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg() == initValue) {
+ Thread.sleep(50 + (i * 10));
+ }
+ }
+ Assert.assertNotEquals(pulsar.getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg(), initValue);
+
+ admin.namespaces().createNamespace(namespace);
+
+ // create producer and topic
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName1).create();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName1).get();
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1).subscriptionName(subName1)
+ .subscribe();
+
+ DispatchRateLimiter subRateLimiter = null;
+ Dispatcher subDispatcher = topic.getSubscription(subName1).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ // (1) Update dispatch rate on cluster-config update
+ Assert.assertEquals(clusterMessageRate, subRateLimiter.getDispatchRateOnMsg());
+
+ // (2) Update namespace throttling limit
+ int nsMessageRate = 500;
+ DispatchRate dispatchRate = new DispatchRate(nsMessageRate, 0, 1);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+ for (int i = 0; i < 5; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() != nsMessageRate) {
+ Thread.sleep(50 + (i * 10));
+ }
+ }
+ Assert.assertEquals(nsMessageRate, subRateLimiter.getDispatchRateOnMsg());
+
+ // (3) Disable namespace throttling limit will force to take cluster-config
+ dispatchRate = new DispatchRate(0, 0, 1);
+ admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
+ for (int i = 0; i < 5; i++) {
+ if (subRateLimiter.getDispatchRateOnMsg() == nsMessageRate) {
+ Thread.sleep(50 + (i * 10));
+ }
+ }
+ Assert.assertEquals(clusterMessageRate, subRateLimiter.getDispatchRateOnMsg());
+
+ // (5) Namespace throttling is disabled so, new topic should take cluster throttling limit
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2).create();
+ PersistentTopic topic2 = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName2).get();
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName2)
+ .subscribe();
+
+ subDispatcher = topic2.getSubscription(subName2).getDispatcher();
+ if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getDispatchRateLimiter();
+ } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
+ subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getDispatchRateLimiter();
+ } else {
+ Assert.fail("Should only have PersistentDispatcher in this test");
+ }
+
+ Assert.assertEquals(clusterMessageRate, subRateLimiter.getDispatchRateOnMsg());
+
+ producer.close();
+ producer2.close();
+
+ log.info("-- Exiting {} test --", methodName);
+ }
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java
index 157d50aaa..9fa0a53a2 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java
@@ -825,6 +825,28 @@
*/
DispatchRate getDispatchRate(String namespace) throws PulsarAdminException;
+ /**
+ * Set subscription-message-dispatch-rate (subscriptions under this namespace can dispatch this many messages per second)
+ *
+ * @param namespace
+ * @param dispatchRate
+ * number of messages per second
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void setSubscriptionDispatchRate(String namespace, DispatchRate dispatchRate) throws PulsarAdminException;
+
+ /** Get subscription-message-dispatch-rate (subscriptions under this namespace can dispatch this many messages per second)
+ *
+ * @param namespace
+ * @returns DispatchRate
+ * number of messages per second
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ DispatchRate getSubscriptionDispatchRate(String namespace) throws PulsarAdminException;
+
+
/**
* Clear backlog for all topics on a namespace
*
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
index a75da4996..c24d8e562 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
@@ -429,6 +429,29 @@ public DispatchRate getDispatchRate(String namespace) throws PulsarAdminExceptio
}
}
+ @Override
+ public void setSubscriptionDispatchRate(String namespace, DispatchRate dispatchRate) throws PulsarAdminException {
+ try {
+ NamespaceName ns = NamespaceName.get(namespace);
+ request(adminNamespaces.path(ns.getProperty()).path(ns.getCluster()).path(ns.getLocalName())
+ .path("subscriptionDispatchRate"))
+ .post(Entity.entity(dispatchRate, MediaType.APPLICATION_JSON), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public DispatchRate getSubscriptionDispatchRate(String namespace) throws PulsarAdminException {
+ try {
+ NamespaceName ns = NamespaceName.get(namespace);
+ return request(adminNamespaces.path(ns.getProperty()).path(ns.getCluster()).path(ns.getLocalName())
+ .path("subscriptionDispatchRate")).get(DispatchRate.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
@Override
public void clearNamespaceBacklog(String namespace) throws PulsarAdminException {
try {
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
index 9d93bca2e..6d37594f9 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
@@ -419,6 +419,43 @@ void run() throws PulsarAdminException {
}
}
+ @Parameters(commandDescription = "Set subscription message-dispatch-rate for all subscription of the namespace")
+ private class SetSuscriptionDispatchRate extends CliCommand {
+ @Parameter(description = "property/cluster/namespace/\n", required = true)
+ private java.util.List<String> params;
+
+ @Parameter(names = { "--sub-msg-dispatch-rate",
+ "-md" }, description = "message-dispatch-rate (default -1 will be overwrite if not passed)\n", required = false)
+ private int msgDispatchRate = -1;
+
+ @Parameter(names = { "--byte-dispatch-rate",
+ "-bd" }, description = "byte-dispatch-rate (default -1 will be overwrite if not passed)\n", required = false)
+ private long byteDispatchRate = -1;
+
+ @Parameter(names = { "--dispatch-rate-period",
+ "-dt" }, description = "dispatch-rate-period in second type (default 1 second will be overwrite if not passed)\n", required = false)
+ private int dispatchRatePeriodSec = 1;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String namespace = validateNamespace(params);
+ admin.namespaces().setSubscriptionDispatchRate(namespace,
+ new DispatchRate(msgDispatchRate, byteDispatchRate, dispatchRatePeriodSec));
+ }
+ }
+
+ @Parameters(commandDescription = "Get subscription configured message-dispatch-rate for all topics of the namespace (Disabled if value < 0)")
+ private class GetSubscriptionDispatchRate extends CliCommand {
+ @Parameter(description = "property/cluster/namespace\n", required = true)
+ private java.util.List<String> params;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String namespace = validateNamespace(params);
+ print(admin.namespaces().getSubscriptionDispatchRate(namespace));
+ }
+ }
+
@Parameters(commandDescription = "Get the backlog quota policies for a namespace")
private class GetBacklogQuotaMap extends CliCommand {
@Parameter(description = "property/cluster/namespace\n", required = true)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java
index 2e568693f..90e4e664e 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java
@@ -24,28 +24,28 @@
public class DispatchRate {
- public int dispatchThrottlingRatePerTopicInMsg = -1;
- public long dispatchThrottlingRatePerTopicInByte = -1;
+ public int dispatchThrottlingRateInMsg = -1;
+ public long dispatchThrottlingRateInByte = -1;
public int ratePeriodInSecond = 1; /* by default dispatch-rate will be calculate per 1 second */
public DispatchRate() {
super();
- this.dispatchThrottlingRatePerTopicInMsg = -1;
- this.dispatchThrottlingRatePerTopicInByte = -1;
+ this.dispatchThrottlingRateInMsg = -1;
+ this.dispatchThrottlingRateInByte = -1;
this.ratePeriodInSecond = 1;
}
- public DispatchRate(int dispatchThrottlingRatePerTopicInMsg, long dispatchThrottlingRatePerTopicInByte,
+ public DispatchRate(int dispatchThrottlingRateInMsg, long dispatchThrottlingRateInByte,
int ratePeriodInSecond) {
super();
- this.dispatchThrottlingRatePerTopicInMsg = dispatchThrottlingRatePerTopicInMsg;
- this.dispatchThrottlingRatePerTopicInByte = dispatchThrottlingRatePerTopicInByte;
+ this.dispatchThrottlingRateInMsg = dispatchThrottlingRateInMsg;
+ this.dispatchThrottlingRateInByte = dispatchThrottlingRateInByte;
this.ratePeriodInSecond = ratePeriodInSecond;
}
@Override
public int hashCode() {
- return Objects.hash(dispatchThrottlingRatePerTopicInMsg, dispatchThrottlingRatePerTopicInByte,
+ return Objects.hash(dispatchThrottlingRateInMsg, dispatchThrottlingRateInByte,
ratePeriodInSecond);
}
@@ -53,8 +53,8 @@ public int hashCode() {
public boolean equals(Object obj) {
if (obj instanceof DispatchRate) {
DispatchRate rate = (DispatchRate) obj;
- return Objects.equals(dispatchThrottlingRatePerTopicInMsg, rate.dispatchThrottlingRatePerTopicInMsg)
- && Objects.equals(dispatchThrottlingRatePerTopicInByte, rate.dispatchThrottlingRatePerTopicInByte)
+ return Objects.equals(dispatchThrottlingRateInMsg, rate.dispatchThrottlingRateInMsg)
+ && Objects.equals(dispatchThrottlingRateInByte, rate.dispatchThrottlingRateInByte)
&& Objects.equals(ratePeriodInSecond, rate.ratePeriodInSecond);
}
return false;
@@ -63,8 +63,8 @@ public boolean equals(Object obj) {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
- .add("dispatchThrottlingRatePerTopicInMsg", dispatchThrottlingRatePerTopicInMsg)
- .add("dispatchThrottlingRatePerTopicInByte", dispatchThrottlingRatePerTopicInByte)
+ .add("dispatchThrottlingRateInMsg", dispatchThrottlingRateInMsg)
+ .add("dispatchThrottlingRateInByte", dispatchThrottlingRateInByte)
.add("ratePeriodInSecond", ratePeriodInSecond).toString();
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
index 03a8ce3d9..985a13f5f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
@@ -33,6 +33,7 @@
public BundlesData bundles = defaultBundle();
public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> backlog_quota_map = Maps.newHashMap();
public Map<String, DispatchRate> clusterDispatchRate = Maps.newHashMap();
+ public Map<String, DispatchRate> subscriptionDispatchRate = Maps.newHashMap();
public PersistencePolicies persistence = null;
// If set, it will override the broker settings for enabling deduplication
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services