From 65d7aaf1d06842707833753ea3ca01688735009f Mon Sep 17 00:00:00 2001 From: rdhabalia Date: Sun, 23 Jul 2017 20:35:24 -0700 Subject: [PATCH 1/2] Introduce message-dispatch rate limiting read messages based on available-message-rate-permits add admin command add message-rate + byte-rate options for dispatch-rate limiting policies on global-zk + combined msg/byte rate + reset rateLimiter-timeUnit --- conf/broker.conf | 8 + conf/standalone.conf | 8 + .../pulsar/broker/ServiceConfiguration.java | 24 + .../pulsar/broker/admin/AdminResource.java | 6 +- .../pulsar/broker/admin/Namespaces.java | 61 ++ .../cache/LocalZooKeeperCacheService.java | 7 +- .../broker/namespace/NamespaceService.java | 1 - .../pulsar/broker/service/BrokerService.java | 40 +- .../pulsar/broker/service/Consumer.java | 45 +- ...PersistentDispatcherMultipleConsumers.java | 2 +- .../persistent/DispatchRateLimiter.java | 232 +++++++ ...PersistentDispatcherMultipleConsumers.java | 41 +- ...sistentDispatcherSingleActiveConsumer.java | 47 +- .../persistent/PersistentSubscription.java | 5 - .../service/persistent/PersistentTopic.java | 18 +- .../auth/MockedPulsarServiceBaseTest.java | 6 +- ...sistentDispatcherFailoverConsumerTest.java | 7 + .../broker/service/PersistentTopicTest.java | 7 + .../pulsar/broker/service/ServerCnxTest.java | 8 + .../api/MessageDispatchThrottlingTest.java | 640 ++++++++++++++++++ .../pulsar/client/admin/Namespaces.java | 24 +- .../client/admin/internal/NamespacesImpl.java | 25 +- .../pulsar/admin/cli/CmdNamespaces.java | 41 ++ .../common/policies/data/DispatchRate.java | 69 ++ .../pulsar/common/policies/data/Policies.java | 4 + .../pulsar/common/util/RateLimiter.java | 260 +++++++ .../pulsar/common/util/RateLimiterTest.java | 165 +++++ site/docs/latest/admin-api/namespaces.md | 56 ++ 28 files changed, 1825 insertions(+), 32 deletions(-) create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java diff --git a/conf/broker.conf b/conf/broker.conf index 72a234e47f1e4..5e2ff92f0af4b 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -106,6 +106,14 @@ maxUnackedMessagesPerBroker=0 # limit/2 messages maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16 +# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default +# message dispatch-throttling +dispatchThrottlingRatePerTopicInMsg=0 + +# Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling +# default message-byte dispatch-throttling +dispatchThrottlingRatePerTopicInByte=0 + # Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic maxConcurrentLookupRequest=10000 diff --git a/conf/standalone.conf b/conf/standalone.conf index 100061d930d75..a121bc2126901 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -95,6 +95,14 @@ maxUnackedMessagesPerBroker=0 # limit/2 messages maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16 +# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default +# message dispatch-throttling +dispatchThrottlingRatePerTopicInMsg=0 + +# Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling +# default message-byte dispatch-throttling +dispatchThrottlingRatePerTopicInByte=0 + # Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic maxConcurrentLookupRequest=10000 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 dedd02d2aaba1..e472e588fe906 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 @@ -105,6 +105,14 @@ public class ServiceConfiguration implements PulsarConfiguration { // than this percentage limit and subscription will not receive any new messages until that subscription acks back // limit/2 messages private double maxUnackedMessagesPerSubscriptionOnBrokerBlocked = 0.16; + // Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default + // message dispatch-throttling + @FieldContext(dynamic = true) + private int dispatchThrottlingRatePerTopicInMsg = 0; + // Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling + // default message-byte dispatch-throttling + @FieldContext(dynamic = true) + private long dispatchThrottlingRatePerTopicInByte = 0; // Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic @FieldContext(dynamic = true) private int maxConcurrentLookupRequest = 10000; @@ -499,6 +507,22 @@ public void setMaxUnackedMessagesPerSubscriptionOnBrokerBlocked( this.maxUnackedMessagesPerSubscriptionOnBrokerBlocked = maxUnackedMessagesPerSubscriptionOnBrokerBlocked; } + public int getDispatchThrottlingRatePerTopicInMsg() { + return dispatchThrottlingRatePerTopicInMsg; + } + + public void setDispatchThrottlingRatePerTopicInMsg(int dispatchThrottlingRatePerTopicInMsg) { + this.dispatchThrottlingRatePerTopicInMsg = dispatchThrottlingRatePerTopicInMsg; + } + + public long getDispatchThrottlingRatePerTopicInByte() { + return dispatchThrottlingRatePerTopicInByte; + } + + public void setDispatchThrottlingRatePerTopicInByte(long dispatchThrottlingRatePerTopicInByte) { + this.dispatchThrottlingRatePerTopicInByte = dispatchThrottlingRatePerTopicInByte; + } + public int getMaxConcurrentLookupRequest() { return maxConcurrentLookupRequest; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index 8929079901fdf..7332eadb02954 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -37,7 +37,6 @@ import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; -import org.apache.pulsar.common.naming.DestinationDomain; import org.apache.pulsar.common.naming.DestinationName; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceBundleFactory; @@ -46,6 +45,7 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.BundlesData; import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.LocalPolicies; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.PropertyAdmin; import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicies; @@ -258,6 +258,10 @@ ZooKeeperDataCache policiesCache() { return pulsar().getConfigurationCache().policiesCache(); } + ZooKeeperDataCache localPoliciesCache() { + return pulsar().getLocalZkCacheService().policiesCache(); + } + ZooKeeperDataCache clustersCache() { return pulsar().getConfigurationCache().clustersCache(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java index cfc14275b9078..be674fc1415ce 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java @@ -67,6 +67,7 @@ import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; import org.apache.pulsar.common.policies.data.BundlesData; import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.DispatchRate; import org.apache.pulsar.common.policies.data.PersistencePolicies; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.RetentionPolicies; @@ -811,6 +812,66 @@ public void splitNamespaceBundle(@PathParam("property") String property, @PathPa } } + @POST + @Path("/{property}/{cluster}/{namespace}/dispatchRate") + @ApiOperation(value = "Set dispatch-rate throttling for all topics of the namespace") + @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") }) + public void setDispatchRate(@PathParam("property") String property, @PathParam("cluster") String cluster, + @PathParam("namespace") String namespace, DispatchRate dispatchRate) { + log.info("[{}] Set namespace dispatch-rate {}/{}/{}/{}", clientAppId(), property, cluster, namespace, + dispatchRate); + validateSuperUserAccess(); + + if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) { + validateClusterOwnership(cluster); + validateClusterForProperty(property, cluster); + } + + Entry policiesNode = null; + NamespaceName nsName = new NamespaceName(property, cluster, namespace); + + try { + // Force to read the data s.t. the watch to the cache content is setup. + policiesNode = policiesCache().getWithStat(path("policies", property, cluster, namespace)) + .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace " + nsName + " does not exist")); + policiesNode.getKey().clusterDispatchRate.put(cluster, dispatchRate); + + // Write back the new policies into zookeeper + globalZk().setData(path("policies", property, cluster, namespace), + jsonMapper().writeValueAsBytes(policiesNode.getKey()), policiesNode.getValue().getVersion()); + policiesCache().invalidate(path("policies", property, cluster, namespace)); + + log.info("[{}] Successfully updated the dispatchRate for cluster on namespace {}/{}/{}", clientAppId(), + property, cluster, namespace); + } catch (KeeperException.NoNodeException e) { + log.warn("[{}] Failed to update the dispatchRate for cluster on namespace {}/{}/{}: does not exist", + clientAppId(), property, cluster, namespace); + throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); + } catch (KeeperException.BadVersionException e) { + log.warn( + "[{}] Failed to update the dispatchRate for cluster on namespace {}/{}/{} expected policy node version={} : concurrent modification", + clientAppId(), property, cluster, namespace, policiesNode.getValue().getVersion()); + + throw new RestException(Status.CONFLICT, "Concurrent modification"); + } catch (Exception e) { + log.error("[{}] Failed to update the dispatchRate for cluster on namespace {}/{}/{}", clientAppId(), property, + cluster, namespace, e); + throw new RestException(e); + } + } + + @GET + @Path("/{property}/{cluster}/{namespace}/dispatchRate") + @ApiOperation(value = "Get 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 getDispatchRate(@PathParam("property") String property, @PathParam("cluster") String cluster, + @PathParam("namespace") String namespace) { + validateAdminAccessOnProperty(property); + Policies policies = getNamespacePolicies(property, cluster, namespace); + return policies.clusterDispatchRate.get(cluster); + } + @GET @Path("/{property}/{cluster}/{namespace}/backlogQuotaMap") @ApiOperation(value = "Get backlog quota map on a namespace.") diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java index def2871c6f2b2..71deae987e2ee 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java @@ -144,10 +144,11 @@ private void initZK() throws PulsarServerException { */ @SuppressWarnings("deprecation") public CompletableFuture> createPolicies(String path, boolean readFromGlobal) { - checkNotNull(path, "path can't be null"); - checkArgument(path.startsWith(LOCAL_POLICIES_ROOT), "Invalid path of local policies"); - CompletableFuture> future = new CompletableFuture<>(); + if (path == null || !path.startsWith(LOCAL_POLICIES_ROOT)) { + future.completeExceptionally(new IllegalArgumentException("Invalid path of local policies " + path)); + return future; + } if (LOG.isDebugEnabled()) { LOG.debug("Creating local namespace policies for {} - readFromGlobal: {}", path, readFromGlobal); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index cb29d557c3cd2..0ff6630cbc0c7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -23,7 +23,6 @@ import static java.lang.String.format; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; -import static org.apache.pulsar.broker.admin.AdminResource.jsonMapper; import static org.apache.pulsar.broker.cache.LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT; import static org.apache.pulsar.broker.web.PulsarWebResource.joinPath; import static org.apache.pulsar.common.naming.NamespaceBundleFactory.getBundlesData; 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 3beb219289beb..5b5eeba6ee910 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 @@ -63,10 +63,12 @@ import org.apache.pulsar.broker.authentication.AuthenticationService; import org.apache.pulsar.broker.authorization.AuthorizationManager; import org.apache.pulsar.broker.loadbalance.LoadManager; +import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException; import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException; 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.PersistentReplicator; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -87,6 +89,7 @@ import org.apache.pulsar.common.naming.NamespaceBundleFactory; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.DispatchRate; import org.apache.pulsar.common.policies.data.PersistencePolicies; import org.apache.pulsar.common.policies.data.PersistentOfflineTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicStats; @@ -119,7 +122,6 @@ import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.util.concurrent.DefaultThreadFactory; -import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException; import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; public class BrokerService implements Closeable, ZooKeeperCacheListener { @@ -1017,9 +1019,45 @@ private void updateConfigurationAndRegisterListeners() { log.warn("Failed to change load manager due to {}", ex); } }); + // add listener to update message-dispatch-rate in msg + registerConfigurationListener("dispatchThrottlingRatePerTopicInMsg", (dispatchRatePerTopicInMsg) -> { + DispatchRate dispatchRate = new DispatchRate((int) dispatchRatePerTopicInMsg, + pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInByte(), 1); + updateTopicMessageDispatchRate(dispatchRate); + }); + // add listener to update message-dispatch-rate in byte + registerConfigurationListener("dispatchThrottlingRatePerTopicInByte", (dispatchRatePerTopicInByte) -> { + DispatchRate dispatchRate = new DispatchRate(pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), + (long) dispatchRatePerTopicInByte, 1); + updateTopicMessageDispatchRate(dispatchRate); + }); // add more listeners here } + private void updateTopicMessageDispatchRate(final DispatchRate dispatchRate) { + this.pulsar().getExecutor().submit(() -> { + // update message-rate for each topic + topics.forEach((name, topicFuture) -> { + if (topicFuture.isDone()) { + String topicName = null; + try { + if (topicFuture.get() instanceof PersistentTopic) { + PersistentTopic topic = (PersistentTopic) topicFuture.get(); + topicName = topicFuture.get().getName(); + // update broker-dispatch throttling only if namespace-policy is not configured + DispatchRateLimiter rateLimiter = topic.getDispatchRateLimiter(); + if (rateLimiter.getPoliciesDispatchRate() == null) { + rateLimiter.updateDispatchRate(dispatchRate); + } + } + } catch (Exception e) { + log.warn("[{}] failed to update message-dispatch rate {}", topicName, dispatchRate); + } + } + }); + }); + } + /** * Allows a listener to listen on update of {@link ServiceConfiguration} change, so listener can take appropriate * action if any specific config-field value has been changed. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 83235f2c688b7..b9f12543768db 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -144,29 +144,31 @@ public String consumerName() { * * @return a promise that can be use to track when all the data has been written into the socket */ - public Pair sendMessages(final List entries) { + public SendMessageInfo sendMessages(final List entries) { final ChannelHandlerContext ctx = cnx.ctx(); - final MutablePair sentMessages = new MutablePair(); + final SendMessageInfo sentMessages = new SendMessageInfo(); final ChannelPromise writePromise = ctx.newPromise(); - sentMessages.setLeft(writePromise); + sentMessages.channelPromse = writePromise; if (entries.isEmpty()) { if (log.isDebugEnabled()) { log.debug("[{}] List of messages is empty, triggering write future immediately for consumerId {}", subscription, consumerId); } writePromise.setSuccess(); - sentMessages.setRight(0); + sentMessages.totalSentMessages = 0; + sentMessages.totalSentMessageBytes = 0; return sentMessages; } try { - sentMessages.setRight(updatePermitsAndPendingAcks(entries)); + updatePermitsAndPendingAcks(entries, sentMessages); } catch (PulsarServerException pe) { log.warn("[{}] [{}] consumer doesn't support batch-message {}", subscription, consumerId, cnx.getRemoteEndpointProtocolVersion()); subscription.markTopicWithBatchMessagePublished(); - sentMessages.setRight(0); + sentMessages.totalSentMessages = 0; + sentMessages.totalSentMessageBytes = 0; // disconnect consumer: it will update dispatcher's availablePermits and resend pendingAck-messages of this // consumer to other consumer disconnect(); @@ -235,7 +237,7 @@ public static int getBatchSizeforEntry(ByteBuf metadataAndPayload, String subscr return -1; } - int updatePermitsAndPendingAcks(final List entries) throws PulsarServerException { + void updatePermitsAndPendingAcks(final List entries, SendMessageInfo sentMessages) throws PulsarServerException { int permitsToReduce = 0; Iterator iter = entries.iterator(); boolean unsupportedVersion = false; @@ -276,7 +278,8 @@ int updatePermitsAndPendingAcks(final List entries) throws PulsarServerEx } msgOut.recordMultipleEvents(permitsToReduce, totalReadableBytes); - return permitsToReduce; + sentMessages.totalSentMessages = permitsToReduce; + sentMessages.totalSentMessageBytes = totalReadableBytes; } public boolean isWritable() { @@ -575,5 +578,31 @@ private void clearUnAckedMsgs(Consumer consumer) { subscription.addUnAckedMessages(-unaAckedMsgs); } + public static class SendMessageInfo { + ChannelPromise channelPromse; + int totalSentMessages; + long totalSentMessageBytes; + + public ChannelPromise getChannelPromse() { + return channelPromse; + } + public void setChannelPromse(ChannelPromise channelPromse) { + this.channelPromse = channelPromse; + } + public int getTotalSentMessages() { + return totalSentMessages; + } + public void setTotalSentMessages(int totalSentMessages) { + this.totalSentMessages = totalSentMessages; + } + public long getTotalSentMessageBytes() { + return totalSentMessageBytes; + } + public void setTotalSentMessageBytes(long totalSentMessageBytes) { + this.totalSentMessageBytes = totalSentMessageBytes; + } + + } + private static final Logger log = LoggerFactory.getLogger(Consumer.class); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java index 2b2bcb3abed84..2b2902b69c68c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java @@ -145,7 +145,7 @@ public SubType getType() { public void sendMessages(List entries) { Consumer consumer = TOTAL_AVAILABLE_PERMITS_UPDATER.get(this) > 0 ? getNextConsumer() : null; if (consumer != null) { - TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -consumer.sendMessages(entries).getRight()); + TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -consumer.sendMessages(entries).getTotalSentMessages()); } else { entries.forEach(entry -> { int totalMsgs = getBatchSizeforEntry(entry.getDataBuffer(), name, -1); 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 new file mode 100644 index 0000000000000..e2d37a52a5cae --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java @@ -0,0 +1,232 @@ +/** + * 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.broker.service.persistent; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.pulsar.broker.web.PulsarWebResource.path; +import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.common.naming.DestinationName; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.policies.data.DispatchRate; +import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.util.RateLimiter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DispatchRateLimiter { + + private final String topicName; + private final BrokerService brokerService; + private RateLimiter dispatchRateLimiterOnMessage; + private RateLimiter dispatchRateLimiterOnByte; + + public DispatchRateLimiter(PersistentTopic topic) { + this.topicName = topic.getName(); + this.brokerService = topic.getBrokerService(); + updateDispatchRate(); + registerLocalPoliciesListener(); + } + + /** + * returns available msg-permit if msg-dispatch-throttling is enabled else it returns -1 + * + * @return + */ + public long getAvailableDispatchRateLimitOnMsg() { + return dispatchRateLimiterOnMessage == null ? -1 : dispatchRateLimiterOnMessage.getAvailablePermits(); + } + + /** + * It acquires msg and bytes permits from rate-limiter and returns if acquired permits succeed. + * + * @param msgPermits + * @param bytePermits + * @return + */ + public boolean tryDispatchPermit(long msgPermits, long bytePermits) { + boolean acquiredMsgPermit = msgPermits <= 0 || dispatchRateLimiterOnMessage == null + // acquiring permits must be < configured msg-rate; + || dispatchRateLimiterOnMessage.tryAcquire(msgPermits); + boolean acquiredBytePermit = bytePermits <= 0 || dispatchRateLimiterOnByte == null + // acquiring permits must be < configured msg-rate; + || dispatchRateLimiterOnByte.tryAcquire(bytePermits); + return acquiredMsgPermit && acquiredBytePermit; + } + + /** + * checks if dispatch-rate limit is configured and if it's configured then check if permits are available or not. + * + * @return + */ + public boolean hasMessageDispatchPermit() { + return (dispatchRateLimiterOnMessage == null || dispatchRateLimiterOnMessage.getAvailablePermits() > 0) + && (dispatchRateLimiterOnByte == null || dispatchRateLimiterOnByte.getAvailablePermits() > 0); + } + + /** + * Checks if dispatch-rate limiting is enabled. + * + * @return + */ + public boolean isDispatchRateLimitingEnabled() { + return dispatchRateLimiterOnMessage != null || dispatchRateLimiterOnByte != null; + } + + /** + * Update dispatch-throttling-rate. gives first priority to namespace-policy configured dispatch rate else applies + * default broker dispatch-throttling-rate + */ + private void updateDispatchRate() { + DispatchRate dispatchRate = getPoliciesDispatchRate(); + if (dispatchRate == null) { + dispatchRate = new DispatchRate(brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), + brokerService.pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInByte(), 1); + } + updateDispatchRate(dispatchRate); + log.info("[{}] configured message-dispatch rate at broker {}", this.topicName, dispatchRate); + } + + /** + * Register listener on namespace policy change to update dispatch-rate if required + * + */ + private void registerLocalPoliciesListener() { + brokerService.pulsar().getConfigurationCache().policiesCache().registerListener((path, data, stat) -> { + final NamespaceName namespace = DestinationName.get(this.topicName).getNamespaceObject(); + final String cluster = brokerService.pulsar().getConfiguration().getClusterName(); + final String policiesPath = path(POLICIES, namespace.toString()); + if (policiesPath.equals(path)) { + DispatchRate dispatchRate = data.clusterDispatchRate.get(cluster); + // update dispatch-rate only if it's configured in policies else ignore + if (dispatchRate != null) { + updateDispatchRate(dispatchRate); + } + } + }); + } + + /** + * Gets configured dispatch-rate from namespace policies. Returns null if dispach-rate is not configured + * + * @return + */ + public DispatchRate getPoliciesDispatchRate() { + final NamespaceName namespace = DestinationName.get(this.topicName).getNamespaceObject(); + final String cluster = brokerService.pulsar().getConfiguration().getClusterName(); + final String path = path(POLICIES, namespace.toString()); + try { + Optional policies = brokerService.pulsar().getConfigurationCache().policiesCache().getAsync(path) + .get(cacheTimeOutInSec, SECONDS); + if (policies.isPresent() && policies.get().clusterDispatchRate != null + && policies.get().clusterDispatchRate.get(cluster) != null) { + return policies.get().clusterDispatchRate.get(cluster); + } + } catch (Exception e) { + log.warn("Failed to get message-rate for {}", this.topicName, e); + } + return null; + } + + /** + * Update dispatch rate by updating msg and byte rate-limiter. If dispatch-rate is configured < 0 then it shutdowns + * 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); + + long msgRate = dispatchRate.dispatchThrottlingRatePerTopicInMsg; + long byteRate = dispatchRate.dispatchThrottlingRatePerTopicInByte; + long ratePerid = dispatchRate.ratePeriodInSecond; + + // update msg-rateLimiter + if (msgRate > 0) { + if (this.dispatchRateLimiterOnMessage == null) { + this.dispatchRateLimiterOnMessage = new RateLimiter(brokerService.pulsar().getExecutor(), msgRate, + ratePerid, TimeUnit.SECONDS); + } else { + this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.ratePeriodInSecond, + TimeUnit.SECONDS); + } + } else { + // message-rate should be disable and close + if (this.dispatchRateLimiterOnMessage != null) { + this.dispatchRateLimiterOnMessage.shutdown(); + this.dispatchRateLimiterOnMessage = null; + } + } + + // update byte-rateLimiter + if (byteRate > 0) { + if (this.dispatchRateLimiterOnByte == null) { + this.dispatchRateLimiterOnByte = new RateLimiter(brokerService.pulsar().getExecutor(), byteRate, + ratePerid, TimeUnit.SECONDS); + } else { + this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.ratePeriodInSecond, + TimeUnit.SECONDS); + } + } else { + // message-rate should be disable and close + if (this.dispatchRateLimiterOnByte != null) { + this.dispatchRateLimiterOnByte.shutdown(); + this.dispatchRateLimiterOnByte = null; + } + } + } + + /** + * Get configured msg dispatch-throttling rate. Returns -1 if not configured + * + * @return + */ + public long getDispatchRateOnMsg() { + return dispatchRateLimiterOnMessage != null ? dispatchRateLimiterOnMessage.getRate() : -1; + } + + /** + * Get configured byte dispatch-throttling rate. Returns -1 if not configured + * + * @return + */ + public long getDispatchRateOnByte() { + return dispatchRateLimiterOnByte != null ? dispatchRateLimiterOnByte.getRate() : -1; + } + + public void close() { + // shutdown rate-limiter + if (dispatchRateLimiterOnMessage != null) { + dispatchRateLimiterOnMessage.shutdown(); + dispatchRateLimiterOnMessage = null; + } + if (dispatchRateLimiterOnByte != null) { + dispatchRateLimiterOnByte.shutdown(); + dispatchRateLimiterOnByte = null; + } + } + + private static final Logger log = LoggerFactory.getLogger(DispatchRateLimiter.class); +} 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 06d22d447cc52..07012914f2b67 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 @@ -18,12 +18,14 @@ */ package org.apache.pulsar.broker.service.persistent; +import static java.util.stream.Collectors.toSet; +import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS; + import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import static java.util.stream.Collectors.toSet; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.Entry; @@ -36,6 +38,7 @@ import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.Consumer.SendMessageInfo; import org.apache.pulsar.broker.service.Dispatcher; import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType; @@ -45,8 +48,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.carrotsearch.hppc.ObjectHashSet; -import com.carrotsearch.hppc.ObjectSet; import com.google.common.collect.ComparisonChain; import com.google.common.collect.Lists; @@ -171,6 +172,31 @@ public void readMoreEntries() { if (totalAvailablePermits > 0 && isAtleastOneConsumerAvailable()) { int messagesToRead = Math.min(totalAvailablePermits, readBatchSize); + // throttle only if: (1) cursor is not active bcz active-cursor reads message from cache rather from + // bookkeeper (2) if topic has reached message-rate threshold: then schedule the read after + // MESSAGE_RATE_BACKOFF_MS + if (!cursor.isActive()) { + DispatchRateLimiter rateLimiter = topic.getDispatchRateLimiter(); + if (rateLimiter.isDispatchRateLimitingEnabled()) { + if (!rateLimiter.hasMessageDispatchPermit()) { + if (log.isDebugEnabled()) { + log.debug("[{}] message-read exceeded message-rate {}/{}, schedule after a {}", name, + rateLimiter.getDispatchRateOnMsg(), rateLimiter.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 = rateLimiter.getAvailableDispatchRateLimitOnMsg(); + if (availablePermitsOnMsg > 0) { + messagesToRead = Math.min(messagesToRead, (int) availablePermitsOnMsg); + } + } + } + } + if (!messagesToReplay.isEmpty()) { if (havePendingReplayRead) { log.debug("[{}] Skipping replay while awaiting previous read to complete", name); @@ -299,6 +325,8 @@ public synchronized void readEntriesComplete(List entries, Object ctx) { log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size()); } + long totalMessagesSent = 0; + long totalBytesSent = 0; while (entriesToDispatch > 0 && totalAvailablePermits > 0 && isAtleastOneConsumerAvailable()) { Consumer c = getNextConsumer(); if (c == null) { @@ -320,14 +348,19 @@ public synchronized void readEntriesComplete(List entries, Object ctx) { }); } - int msgSent = c.sendMessages(entries.subList(start, start + messagesForC)).getRight(); + SendMessageInfo sentMsgInfo = c.sendMessages(entries.subList(start, start + messagesForC)); + long msgSent = sentMsgInfo.getTotalSentMessages(); start += messagesForC; entriesToDispatch -= messagesForC; totalAvailablePermits -= msgSent; + totalMessagesSent += sentMsgInfo.getTotalSentMessageBytes(); + totalBytesSent += sentMsgInfo.getTotalSentMessages(); } } + topic.getDispatchRateLimiter().tryDispatchPermit(totalMessagesSent, totalBytesSent); + if (entriesToDispatch > 0) { if (log.isDebugEnabled()) { log.debug("[{}] No consumers found with available permits, storing {} positions for later replay", name, 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 a233bf7aabc4d..a09beebe027fc 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 @@ -19,10 +19,9 @@ package org.apache.pulsar.broker.service.persistent; import static com.google.common.base.Preconditions.checkNotNull; +import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS; import java.util.List; -import java.util.Objects; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; @@ -34,10 +33,10 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.Consumer.SendMessageInfo; import org.apache.pulsar.broker.service.Dispatcher; import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType; -import org.apache.pulsar.utils.CopyOnWriteArrayList; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -110,10 +109,15 @@ public synchronized void readEntriesComplete(final List entries, Object o readMoreEntries(currentConsumer); } } else { - currentConsumer.sendMessages(entries).getLeft().addListener(future -> { + SendMessageInfo sentMsgInfo = currentConsumer.sendMessages(entries); + final long totalMessagesSent = sentMsgInfo.getTotalSentMessages(); + final long totalBytesSent = sentMsgInfo.getTotalSentMessageBytes(); + sentMsgInfo.getChannelPromse().addListener(future -> { if (future.isSuccess()) { // Schedule a new read batch operation only after the previous batch has been written to the socket synchronized (PersistentDispatcherSingleActiveConsumer.this) { + // acquire message-dispatch permits for already delivered messages + topic.getDispatchRateLimiter().tryDispatchPermit(totalMessagesSent, totalBytesSent); Consumer newConsumer = ACTIVE_CONSUMER_UPDATER.get(this); if (newConsumer != null && !havePendingRead) { readMoreEntries(newConsumer); @@ -193,6 +197,41 @@ protected void readMoreEntries(Consumer consumer) { } int messagesToRead = Math.min(availablePermits, readBatchSize); + + // throttle only if: (1) cursor is not active bcz active-cursor reads message from cache rather from + // bookkeeper (2) if topic has reached message-rate threshold: then schedule the read after + // MESSAGE_RATE_BACKOFF_MS + if (!cursor.isActive()) { + DispatchRateLimiter rateLimiter = topic.getDispatchRateLimiter(); + if (rateLimiter.isDispatchRateLimitingEnabled()) { + if (!rateLimiter.hasMessageDispatchPermit()) { + if (log.isDebugEnabled()) { + log.debug("[{}] message-read exceeded message-rate {}/{}, schedule after a {}", + topic.getName(), rateLimiter.getDispatchRateOnMsg(), rateLimiter.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.info("[{}] 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 availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg(); + if (availablePermitsOnMsg > 0) { + messagesToRead = Math.min(messagesToRead, (int) availablePermitsOnMsg); + } + + } + } + } // Schedule read if (log.isDebugEnabled()) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index fdb076fd05b0b..40825ec677fdb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -133,7 +133,6 @@ public synchronized void addConsumer(Consumer consumer) throws BrokerServiceExce } dispatcher.addConsumer(consumer); - activateCursor(); } @Override @@ -164,10 +163,6 @@ public void deactivateCursor() { this.cursor.setInactive(); } - public void activateCursor() { - this.cursor.setActive(); - } - @Override public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) { dispatcher.consumerFlow(consumer, additionalNumberOfMessages); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 1bca29ddd63ad..f8dc0330be867 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -19,7 +19,11 @@ package org.apache.pulsar.broker.service.persistent; import static com.google.common.base.Preconditions.checkArgument; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.commons.lang3.StringUtils.isBlank; +import static org.apache.pulsar.broker.cache.LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT; +import static org.apache.pulsar.broker.web.PulsarWebResource.path; +import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec; import java.time.Instant; import java.time.ZoneId; @@ -52,6 +56,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.admin.AdminResource; +import org.apache.pulsar.broker.cache.ConfigurationCacheService; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; @@ -148,6 +153,8 @@ public class PersistentTopic implements Topic, AddEntryCallback { // Flag to signal that producer of this topic has published batch-message so, broker should not allow consumer which // doesn't support batch-message private volatile boolean hasBatchMessagePublished = false; + private DispatchRateLimiter dispatchRateLimiter; + public static final int MESSAGE_RATE_BACKOFF_MS = 1000; private static final FastThreadLocal threadLocalTopicStats = new FastThreadLocal() { @Override @@ -189,7 +196,9 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS this.isFenced = false; this.replicatorPrefix = brokerService.pulsar().getConfiguration().getReplicatorPrefix(); USAGE_COUNT_UPDATER.set(this, 0); - + + this.dispatchRateLimiter = new DispatchRateLimiter(this); + for (ManagedCursor cursor : ledger.getCursors()) { if (cursor.getName().startsWith(replicatorPrefix)) { String localCluster = brokerService.pulsar().getConfiguration().getClusterName(); @@ -661,6 +670,9 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { closeFuture.complete(null); } }, null); + + dispatchRateLimiter.close(); + }).exceptionally(exception -> { log.error("[{}] Error closing topic", topic, exception); isFenced = false; @@ -1405,5 +1417,9 @@ public void markBatchMessagePublished() { this.hasBatchMessagePublished = true; } + public DispatchRateLimiter getDispatchRateLimiter() { + return this.dispatchRateLimiter; + } + private static final Logger log = LoggerFactory.getLogger(PersistentTopic.class); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 4a4c965c9939b..5b0900c78604f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -59,7 +59,7 @@ */ public abstract class MockedPulsarServiceBaseTest { - protected final ServiceConfiguration conf; + protected ServiceConfiguration conf; protected PulsarService pulsar; protected PulsarAdmin admin; protected PulsarClient pulsarClient; @@ -78,6 +78,10 @@ public abstract class MockedPulsarServiceBaseTest { private SameThreadOrderedSafeExecutor sameThreadOrderedSafeExecutor; public MockedPulsarServiceBaseTest() { + resetConfig(); + } + + protected void resetConfig() { this.conf = new ServiceConfiguration(); this.conf.setBrokerServicePort(BROKER_PORT); this.conf.setBrokerServicePortTls(BROKER_PORT_TLS); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index 920cfda3b1761..eb3c7028ae774 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -35,6 +35,8 @@ import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; @@ -52,6 +54,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.cache.ConfigurationCacheService; +import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.Consumer; @@ -101,8 +104,12 @@ public void setup() throws Exception { configCacheService = mock(ConfigurationCacheService.class); @SuppressWarnings("unchecked") ZooKeeperDataCache zkDataCache = mock(ZooKeeperDataCache.class); + LocalZooKeeperCacheService zkCache = mock(LocalZooKeeperCacheService.class); + doReturn(CompletableFuture.completedFuture(Optional.empty())).when(zkDataCache).getAsync(any()); + doReturn(zkDataCache).when(zkCache).policiesCache(); doReturn(zkDataCache).when(configCacheService).policiesCache(); doReturn(configCacheService).when(pulsar).getConfigurationCache(); + doReturn(zkCache).when(pulsar).getLocalZkCacheService(); brokerService = spy(new BrokerService(pulsar)); doReturn(brokerService).when(pulsar).getBrokerService(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index 37e9b11020c9a..be14c4219ded5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -72,6 +72,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.cache.ConfigurationCacheService; +import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.nonpersistent.NonPersistentReplicator; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; @@ -141,6 +142,12 @@ public void setup() throws Exception { doReturn(zkDataCache).when(configCacheService).policiesCache(); doReturn(configCacheService).when(pulsar).getConfigurationCache(); doReturn(Optional.empty()).when(zkDataCache).get(anyString()); + + LocalZooKeeperCacheService zkCache = mock(LocalZooKeeperCacheService.class); + doReturn(CompletableFuture.completedFuture(Optional.empty())).when(zkDataCache).getAsync(any()); + doReturn(zkDataCache).when(zkCache).policiesCache(); + doReturn(configCacheService).when(pulsar).getConfigurationCache(); + doReturn(zkCache).when(pulsar).getLocalZkCacheService(); brokerService = spy(new BrokerService(pulsar)); doReturn(brokerService).when(pulsar).getBrokerService(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index 653d10b81efa4..33743a2062e0a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -60,6 +60,7 @@ import org.apache.pulsar.broker.authentication.AuthenticationService; import org.apache.pulsar.broker.authorization.AuthorizationManager; import org.apache.pulsar.broker.cache.ConfigurationCacheService; +import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.ServerCnx; @@ -149,6 +150,13 @@ public void setup() throws Exception { doReturn(Optional.empty()).when(zkDataCache).get(anyObject()); doReturn(zkDataCache).when(configCacheService).policiesCache(); doReturn(configCacheService).when(pulsar).getConfigurationCache(); + + LocalZooKeeperCacheService zkCache = mock(LocalZooKeeperCacheService.class); + doReturn(CompletableFuture.completedFuture(Optional.empty())).when(zkDataCache).getAsync(any()); + doReturn(zkDataCache).when(zkCache).policiesCache(); + doReturn(configCacheService).when(pulsar).getConfigurationCache(); + doReturn(zkCache).when(pulsar).getLocalZkCacheService(); + brokerService = spy(new BrokerService(pulsar)); doReturn(brokerService).when(pulsar).getBrokerService(); 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 new file mode 100644 index 0000000000000..17f19e3883124 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java @@ -0,0 +1,640 @@ +/** + * 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.lang.reflect.Field; +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.pulsar.broker.service.BrokerService; +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.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +public class MessageDispatchThrottlingTest extends ProducerConsumerBase { + private static final Logger log = LoggerFactory.getLogger(MessageDispatchThrottlingTest.class); + + @BeforeMethod + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + this.conf.setClusterName("use"); + } + + @AfterMethod + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + super.resetConfig(); + } + + @DataProvider(name = "subscriptions") + public Object[][] subscriptionsProvider() { + return new Object[][] { new Object[] { SubscriptionType.Shared }, { SubscriptionType.Exclusive } }; + } + + @DataProvider(name = "dispatchRateType") + public Object[][] dispatchRateProvider() { + return new Object[][] { { DispatchRateType.messageRate }, { DispatchRateType.byteRate } }; + } + + @DataProvider(name = "subscriptionAndDispatchRateType") + public Object[][] subDisTypeProvider() { + List mergeList = new LinkedList(); + for (Object[] sub : subscriptionsProvider()) { + for (Object[] dispatch : dispatchRateProvider()) { + mergeList.add(merge(sub, dispatch)); + } + } + return mergeList.toArray(new Object[0][0]); + } + + public static T[] merge(T[] first, T[] last) { + int totalLength = first.length + last.length; + T[] result = Arrays.copyOf(first, totalLength); + int offset = first.length; + System.arraycopy(last, 0, result, offset, first.length); + return result; + } + + enum DispatchRateType { + messageRate, byteRate; + } + + /** + * verifies: message-rate change gets reflected immediately into topic at runtime + * + * @throws Exception + */ + @Test + public void testMessageRateDynamicallyChange() throws Exception { + + log.info("-- Starting {} test --", methodName); + + final String namespace = "my-property/use/throttling_ns"; + final String topicName = "persistent://" + namespace + "/throttlingBlock"; + + admin.namespaces().createNamespace(namespace); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + // (1) verify message-rate is -1 initially + Assert.assertEquals(topic.getDispatchRateLimiter().getDispatchRateOnMsg(), -1); + + // (1) change to 100 + int messageRate = 100; + DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 1); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + boolean isDispatchRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnMsg() > 0) { + isDispatchRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isDispatchRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + // (1) change to 500 + messageRate = 500; + dispatchRate = new DispatchRate(-1, messageRate, 1); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + isDispatchRateUpdate = false; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnByte() == messageRate) { + isDispatchRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isDispatchRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + producer.close(); + } + + /** + * 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 int messageRate = 100; + DispatchRate dispatchRate = null; + if (DispatchRateType.messageRate.equals(dispatchRateType)) { + dispatchRate = new DispatchRate(messageRate, -1, 1); + } else { + dispatchRate = new DispatchRate(-1, messageRate, 1); + } + + admin.namespaces().createNamespace(namespace); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + boolean isMessageRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnMsg() > 0 + || topic.getDispatchRateLimiter().getDispatchRateOnByte() > 0) { + isMessageRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isMessageRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + int numMessages = 500; + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(subscription); + conf.setMessageListener((consumer, 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 consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // 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); + } + + /** + * It verifies that dispatch-rate throttling with cluster-configuration + * + * @param subscription + * @param dispatchRateType + * @throws Exception + */ + @Test() + public void testClusterMsgByteRateLimitingClusterConfig() throws Exception { + log.info("-- Starting {} test --", methodName); + + final String namespace = "my-property/use/throttling_ns"; + final String topicName = "persistent://" + namespace + "/throttlingBlock"; + final int messageRate = 100; + final long byteRate = 1024 * 1024;// 1MB rate enough to let all msg to be delivered + + int initValue = pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg(); + // (1) Update message-dispatch-rate limit + admin.brokers().updateDynamicConfiguration("dispatchThrottlingRatePerTopicInMsg", Integer.toString(messageRate)); + admin.brokers().updateDynamicConfiguration("dispatchThrottlingRatePerTopicInByte", Long.toString(byteRate)); + // sleep incrementally as zk-watch notification is async and may take some time + for (int i = 0; i < 5; i++) { + if (pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg() != initValue) { + Thread.sleep(50 + (i * 10)); + } + } + Assert.assertNotEquals(pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), initValue); + + admin.namespaces().createNamespace(namespace); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + int numMessages = 500; + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(SubscriptionType.Shared); + conf.setMessageListener((consumer, 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 consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // Asynchronously produce messages + for (int i = 0; i < numMessages; i++) { + final String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + // consumer should not have received all published message due to message-rate throttling + Assert.assertTrue(totalReceived.get() < messageRate * 2); + + consumer.close(); + producer.close(); + pulsar.getConfiguration().setDispatchThrottlingRatePerTopicInMsg(initValue); + log.info("-- Exiting {} test --", methodName); + } + + /** + * verify rate-limiting should throttle message-dispatching based on message-rate + * + * 
+     *  1. dispatch-msg-rate = 10 msg/sec
+     *  2. send 20 msgs 
+     *  3. it should take up to 2 second to receive all messages
+     *
+ * + * @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 int messageRate = 10; + DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 1); + admin.namespaces().createNamespace(namespace); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + boolean isMessageRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnMsg() > 0) { + isMessageRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isMessageRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + final int numProducedMessages = 20; + final CountDownLatch latch = new CountDownLatch(numProducedMessages); + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(subscription); + conf.setMessageListener((consumer, 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(); + }); + Consumer consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // 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); + + consumer.close(); + producer.close(); + log.info("-- Exiting {} test --", methodName); + } + + /** + * verify rate-limiting should throttle message-dispatching based on byte-rate + * + * 
+     *  1. dispatch-byte-rate = 100 bytes/sec
+     *  2. send 20 msgs : each with 10 byte
+     *  3. it should take up to 2 second to receive all messages
+     *
+ * + * @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 int byteRate = 100; + DispatchRate dispatchRate = new DispatchRate(-1, byteRate, 1); + admin.namespaces().createNamespace(namespace); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + boolean isMessageRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnByte() > 0) { + isMessageRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isMessageRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + final int numProducedMessages = 20; + final CountDownLatch latch = new CountDownLatch(numProducedMessages); + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(subscription); + conf.setMessageListener((consumer, 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(); + }); + Consumer consumer = pulsarClient.subscribe(topicName, "my-subscriber-name-" + subscription, conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // Asynchronously produce messages + for (int i = 0; i < numProducedMessages; i++) { + producer.send(new byte[byteRate / 10]); + } + + latch.await(); + Assert.assertEquals(totalReceived.get(), numProducedMessages); + + 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 int messageRate = 100; + DispatchRate dispatchRate = new DispatchRate(messageRate, -1, 1); + admin.namespaces().createNamespace(namespace); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + boolean isMessageRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnMsg() > 0) { + isMessageRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isMessageRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + final int numProducedMessages = 500; + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(SubscriptionType.Shared); + conf.setMessageListener((consumer, 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 consumer1 = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + Consumer consumer2 = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + Consumer consumer3 = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + Consumer consumer4 = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + Consumer consumer5 = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // Asynchronously produce messages + for (int i = 0; i < numProducedMessages; i++) { + final String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + // consumer should not have received all published message due to message-rate throttling + Assert.assertTrue(totalReceived.get() < messageRate * 2); + + 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 int messageRate = 100; + + int initValue = pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg(); + // (1) Update message-dispatch-rate limit + admin.brokers().updateDynamicConfiguration("dispatchThrottlingRatePerTopicInMsg", 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().getDispatchThrottlingRatePerTopicInMsg() != initValue) { + Thread.sleep(50 + (i * 10)); + } + } + Assert.assertNotEquals(pulsar.getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), initValue); + + admin.namespaces().createNamespace(namespace); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + int numMessages = 500; + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(subscription); + conf.setMessageListener((consumer, 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 consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + + // Asynchronously produce messages + for (int i = 0; i < numMessages; i++) { + final String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + // consumer should not have received all publihsed message due to message-rate throttling + Assert.assertTrue(totalReceived.get() < messageRate * 2); + + consumer.close(); + producer.close(); + pulsar.getConfiguration().setDispatchThrottlingRatePerTopicInMsg(initValue); + log.info("-- Exiting {} test --", methodName); + } + + /** + * It verifies that that dispatch-throttling considers both msg/byte rate if both of them are configured together + * + * @param subscription + * @throws Exception + */ + @Test(dataProvider = "subscriptions", timeOut = 5000) + public void testMessageByteRateThrottlingCombined(SubscriptionType subscription) throws Exception { + log.info("-- Starting {} test --", methodName); + + final String namespace = "my-property/use/throttling_ns"; + final String topicName = "persistent://" + namespace + "/throttlingAll"; + + final int messageRate = 100; // 100 msgs per second + final long byteRate = 100; // 100 bytes per second + DispatchRate dispatchRate = new DispatchRate(messageRate, byteRate, 1); + admin.namespaces().createNamespace(namespace); + admin.namespaces().setDispatchRate(namespace, dispatchRate); + // create producer and topic + Producer producer = pulsarClient.createProducer(topicName); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName).get(); + boolean isMessageRateUpdate = false; + int retry = 5; + for (int i = 0; i < retry; i++) { + if (topic.getDispatchRateLimiter().getDispatchRateOnMsg() > 0 + && topic.getDispatchRateLimiter().getDispatchRateOnByte() > 0) { + isMessageRateUpdate = true; + break; + } else { + if (i != retry - 1) { + Thread.sleep(100); + } + } + } + Assert.assertTrue(isMessageRateUpdate); + Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate); + + final int numProducedMessages = 200; + + final AtomicInteger totalReceived = new AtomicInteger(0); + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(subscription); + conf.setMessageListener((consumer, 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 consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + // deactive cursors + deactiveCursors((ManagedLedgerImpl) topic.getManagedLedger()); + consumer.close(); + + // Asynchronously produce messages + final int dataSize = 50; + final byte[] data = new byte[dataSize]; + for (int i = 0; i < numProducedMessages; i++) { + producer.send(data); + } + + consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", conf); + final int totalReceivedBytes = dataSize * totalReceived.get(); + Assert.assertNotEquals(totalReceivedBytes, byteRate * 2); + + consumer.close(); + producer.close(); + log.info("-- Exiting {} test --", methodName); + } + + private void deactiveCursors(ManagedLedgerImpl ledger) throws Exception { + Field statsUpdaterField = BrokerService.class.getDeclaredField("statsUpdater"); + statsUpdaterField.setAccessible(true); + ScheduledExecutorService statsUpdater = (ScheduledExecutorService) statsUpdaterField + .get(pulsar.getBrokerService()); + statsUpdater.shutdownNow(); + ledger.getCursors().forEach(cursor -> { + ledger.deactivateCursor(cursor); + }); + } + +} \ No newline at end of file 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 bf600176b4bc5..c44118be60214 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 @@ -29,6 +29,7 @@ import org.apache.pulsar.common.policies.data.AuthAction; import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.BundlesData; +import org.apache.pulsar.common.policies.data.DispatchRate; import org.apache.pulsar.common.policies.data.PersistencePolicies; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.RetentionPolicies; @@ -697,11 +698,32 @@ public interface Namespaces { * Split namespace bundle * * @param namespace - * @bundle range of bundle to split + * @param range of bundle to split * @throws PulsarAdminException * Unexpected error */ void splitNamespaceBundle(String namespace, String bundle) throws PulsarAdminException; + + /** + * Set message-dispatch-rate (topics under this namespace can dispatch this many messages per second) + * + * @param namespace + * @param messageRate + * number of messages per second + * @throws PulsarAdminException + * Unexpected error + */ + void setDispatchRate(String namespace, DispatchRate dispatchRate) throws PulsarAdminException; + + /** Get message-dispatch-rate (topics under this namespace can dispatch this many messages per second) + * + * @param namespace + * @returns messageRate + * number of messages per second + * @throws PulsarAdminException + * Unexpected error + */ + DispatchRate getDispatchRate(String namespace) throws PulsarAdminException; /** * Clear backlog for all destinations 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 541e7c673a042..efb7a777e7349 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 @@ -33,12 +33,13 @@ import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.policies.data.AuthAction; import org.apache.pulsar.common.policies.data.BacklogQuota; +import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; import org.apache.pulsar.common.policies.data.BundlesData; +import org.apache.pulsar.common.policies.data.DispatchRate; import org.apache.pulsar.common.policies.data.ErrorData; import org.apache.pulsar.common.policies.data.PersistencePolicies; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.RetentionPolicies; -import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; public class NamespacesImpl extends BaseResource implements Namespaces { @@ -352,6 +353,28 @@ public void splitNamespaceBundle(String namespace, String bundle) throws PulsarA throw getApiException(e); } } + + @Override + public void setDispatchRate(String namespace, DispatchRate dispatchRate) throws PulsarAdminException { + try { + NamespaceName ns = new NamespaceName(namespace); + request(namespaces.path(ns.getProperty()).path(ns.getCluster()).path(ns.getLocalName()).path("dispatchRate")) + .post(Entity.entity(dispatchRate, MediaType.APPLICATION_JSON), ErrorData.class); + } catch (Exception e) { + throw getApiException(e); + } + } + + @Override + public DispatchRate getDispatchRate(String namespace) throws PulsarAdminException { + try { + NamespaceName ns = new NamespaceName(namespace); + return request(namespaces.path(ns.getProperty()).path(ns.getCluster()).path(ns.getLocalName()).path("dispatchRate")) + .get(DispatchRate.class); + } catch (Exception e) { + throw getApiException(e); + } + } @Override public void clearNamespaceBacklog(String namespace) throws PulsarAdminException { 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 dfdbb48971ebb..506d0756b5892 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 @@ -26,6 +26,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.policies.data.BacklogQuota; +import org.apache.pulsar.common.policies.data.DispatchRate; import org.apache.pulsar.common.policies.data.PersistencePolicies; import org.apache.pulsar.common.policies.data.RetentionPolicies; @@ -293,6 +294,43 @@ void run() throws PulsarAdminException { } } + @Parameters(commandDescription = "Set message-dispatch-rate for all topics of the namespace") + private class SetDispatchRate extends CliCommand { + @Parameter(description = "property/cluster/namespace/\n", required = true) + private java.util.List params; + + @Parameter(names = { "--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().setDispatchRate(namespace, + new DispatchRate(msgDispatchRate, byteDispatchRate, dispatchRatePeriodSec)); + } + } + + @Parameters(commandDescription = "Get configured message-dispatch-rate for all topics of the namespace (Disabled if value < 0)") + private class GetDispatchRate extends CliCommand { + @Parameter(description = "property/cluster/namespace\n", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String namespace = validateNamespace(params); + print(admin.namespaces().getDispatchRate(namespace)); + } + } + @Parameters(commandDescription = "Get the backlog quota policies for a namespace") private class GetBacklogQuotaMap extends CliCommand { @Parameter(description = "property/cluster/namespace\n", required = true) @@ -526,6 +564,9 @@ public CmdNamespaces(PulsarAdmin admin) { jcommander.addCommand("unload", new Unload()); jcommander.addCommand("split-bundle", new SplitBundle()); + + jcommander.addCommand("set-dispatch-rate", new SetDispatchRate()); + jcommander.addCommand("get-dispatch-rate", new GetDispatchRate()); jcommander.addCommand("clear-backlog", new ClearBacklog()); 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 new file mode 100644 index 0000000000000..c7b1a47f509b7 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DispatchRate.java @@ -0,0 +1,69 @@ +/** + * 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 com.google.common.base.Objects; + +public class DispatchRate { + + public int dispatchThrottlingRatePerTopicInMsg = -1; + public long dispatchThrottlingRatePerTopicInByte = -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.ratePeriodInSecond = 1; + } + + public DispatchRate(int dispatchThrottlingRatePerTopicInMsg, long dispatchThrottlingRatePerTopicInByte, + int ratePeriodInSecond) { + super(); + this.dispatchThrottlingRatePerTopicInMsg = dispatchThrottlingRatePerTopicInMsg; + this.dispatchThrottlingRatePerTopicInByte = dispatchThrottlingRatePerTopicInByte; + this.ratePeriodInSecond = ratePeriodInSecond; + } + + @Override + public int hashCode() { + return Objects.hashCode(dispatchThrottlingRatePerTopicInMsg, dispatchThrottlingRatePerTopicInByte, + ratePeriodInSecond); + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof DispatchRate) { + DispatchRate rate = (DispatchRate) obj; + return Objects.equal(dispatchThrottlingRatePerTopicInMsg, rate.dispatchThrottlingRatePerTopicInMsg) + && Objects.equal(dispatchThrottlingRatePerTopicInByte, rate.dispatchThrottlingRatePerTopicInByte) + && Objects.equal(ratePeriodInSecond, rate.ratePeriodInSecond); + } + return false; + } + + @Override + public String toString() { + return Objects.toStringHelper(this) + .add("dispatchThrottlingRatePerTopicInMsg", dispatchThrottlingRatePerTopicInMsg) + .add("dispatchThrottlingRatePerTopicInByte", dispatchThrottlingRatePerTopicInByte) + .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 e3fb3f38fd708..6dd05d6a639d0 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 @@ -31,6 +31,7 @@ public class Policies { public List replication_clusters; public BundlesData bundles; public Map backlog_quota_map; + public Map clusterDispatchRate; public PersistencePolicies persistence; public Map latency_stats_sample_rate; public int message_ttl_in_seconds; @@ -45,6 +46,7 @@ public Policies() { replication_clusters = Lists.newArrayList(); bundles = defaultBundle(); backlog_quota_map = Maps.newHashMap(); + clusterDispatchRate = Maps.newHashMap(); persistence = null; latency_stats_sample_rate = Maps.newHashMap(); message_ttl_in_seconds = 0; @@ -59,6 +61,7 @@ public boolean equals(Object obj) { return Objects.equal(auth_policies, other.auth_policies) && Objects.equal(replication_clusters, other.replication_clusters) && Objects.equal(backlog_quota_map, other.backlog_quota_map) + && Objects.equal(clusterDispatchRate, other.clusterDispatchRate) && Objects.equal(persistence, other.persistence) && Objects.equal(bundles, other.bundles) && Objects.equal(latency_stats_sample_rate, other.latency_stats_sample_rate) && message_ttl_in_seconds == other.message_ttl_in_seconds @@ -82,6 +85,7 @@ public String toString() { return Objects.toStringHelper(this).add("auth_policies", auth_policies) .add("replication_clusters", replication_clusters).add("bundles", bundles) .add("backlog_quota_map", backlog_quota_map).add("persistence", persistence) + .add("clusterDispatchRate", clusterDispatchRate) .add("latency_stats_sample_rate", latency_stats_sample_rate) .add("message_ttl_in_seconds", message_ttl_in_seconds).add("retention_policies", retention_policies) .add("deleted", deleted).toString(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java new file mode 100644 index 0000000000000..25082dd62bc94 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java @@ -0,0 +1,260 @@ +/** + * 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.util; + +import static com.google.common.base.Preconditions.checkArgument; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import com.google.common.base.Objects; + +/** + * + * A Rate Limiter that distributes permits at a configurable rate. Each {@link #acquire()} blocks if necessary until a + * permit is available, and then takes it. Each {@link #tryAcquire()} tries to acquire permits from available permits, + * it returns true if it succeed else returns false. Rate limiter release configured permits at every configured rate + * time, so, on next ticket new fresh permits will be available. + *

+ * For example: if RateLimiter is configured to release 10 permits at every 1 second then RateLimiter will allow to + * acquire 10 permits at any time with in that 1 second. + *

+ * comparison with other RateLimiter such as {@link com.google.common.util.concurrent.RateLimiter} + *

+ *
    + *
  • Per second rate-limiting: Per second rate-limiting not satisfied by Guava-RateLimiter + *

    + * Guava RateLimiter: For X permits: it releases X/1000 permits every msec. therefore, for permits=2/sec => it + * release 1st permit on first 500msec and 2nd permit on next 500ms. therfore, if 2 request comes with in 500msec + * duration then 2nd request fails to acquire permit though we have configured 2 permits/second. + *

    + * RateLimiter: it releases X permits every second. so, in above usecase: if 2 requests comes at the same time + * then both will acquire the permit. + *

  • Faster: RateLimiter is light-weight and faster than Guava-RateLimiter
    • + *
+ * + * + */ +public class RateLimiter { + + private final ScheduledExecutorService executorService; + private long rateTime; + private TimeUnit timeUnit; + private final boolean externalExecutor; + private ScheduledFuture renewTask; + private long permits; + private long acquiredPermits; + private boolean isShutdown; + + public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit) { + this(null, permits, rateTime, timeUnit); + } + + public RateLimiter(final ScheduledExecutorService service, final long permits, final long rateTime, + final TimeUnit timeUnit) { + checkArgument(permits > 0, "rate must be > 0"); + checkArgument(rateTime > 0, "Renew permit time must be > 0"); + + this.rateTime = rateTime; + this.timeUnit = timeUnit; + this.permits = permits; + + if (service != null) { + this.executorService = service; + this.externalExecutor = true; + } else { + final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); + executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + this.executorService = executor; + this.externalExecutor = false; + } + + } + + public synchronized void shutdown() { + if (!isShutdown) { + if (!externalExecutor) { + executorService.shutdownNow(); + } + if (renewTask != null) { + renewTask.cancel(false); + } + isShutdown = true; + } + } + + public synchronized boolean isShutdown() { + return isShutdown; + } + + /** + * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request be granted. + * + * This method is equivalent to {@code acquire(1)}. + * + * @param permits + * the number of permits to acquire + */ + public synchronized void acquire() throws InterruptedException { + acquire(1); + } + + /** + * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request be granted. + * + * @param permits + * the number of permits to acquire + */ + public synchronized void acquire(long acquirePermit) throws InterruptedException { + checkArgument(!isShutdown(), "Rate limiter is already shutdown"); + checkArgument(acquirePermit <= this.permits, + "acquiring permits must be less or equal than initialized rate =" + this.permits); + + // lazy init and start task only once application start using it + if (renewTask == null) { + renewTask = createTask(); + } + + boolean canAcquire = false; + do { + canAcquire = acquirePermit < 0 || acquiredPermits < this.permits; + if (!canAcquire) { + wait(); + } else { + acquiredPermits += acquirePermit; + } + } while (!canAcquire); + } + + /** + * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. + * + *

+ * This method is equivalent to {@code tryAcquire(1)}. + * + * @param permits + * the number of permits to acquire + * @return {@code true} if the permits were acquired, {@code false} otherwise + */ + public synchronized boolean tryAcquire() { + return tryAcquire(1); + } + + /** + * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. + * + * @param permits + * the number of permits to acquire + * @return {@code true} if the permits were acquired, {@code false} otherwise + */ + public synchronized boolean tryAcquire(long acquirePermit) { + checkArgument(!isShutdown(), "Rate limiter is already shutdown"); + // lazy init and start task only once application start using it + if (renewTask == null) { + renewTask = createTask(); + } + + // acquired-permits can't be larger than the rate + if (acquirePermit > this.permits) { + acquiredPermits = this.permits; + return false; + } + boolean canAcquire = acquirePermit < 0 || acquiredPermits < this.permits; + if (canAcquire) { + acquiredPermits += acquirePermit; + } + return canAcquire; + } + + /** + * Return available permits for this {@link RateLimiter} + * + * @return returns 0 if permis is not available + */ + public synchronized long getAvailablePermits() { + return Math.max(0, this.permits - this.acquiredPermits); + } + + /** + * Resets new rate by configuring new value for permits per configured rate-period + * + * @param permits + */ + public synchronized void setRate(long permits) { + this.permits = permits; + } + + /** + * Resets new rate with new permits and rate-time. + * + * @param permits + * @param rateTime + * @param timeUnit + */ + public synchronized void setRate(long permits, long rateTime, TimeUnit timeUnit) { + if(renewTask != null) { + renewTask.cancel(false); + } + this.permits = permits; + this.rateTime = rateTime; + this.timeUnit = timeUnit; + this.renewTask = createTask(); + } + + /** + * Returns configured permit rate per pre-configured rate-period. + * + * @return rate + */ + public synchronized long getRate() { + return this.permits; + } + + public synchronized long getRateTime() { + return this.rateTime; + } + + public synchronized TimeUnit getRateTimeUnit() { + return this.timeUnit; + } + + protected ScheduledFuture createTask() { + return executorService.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + renew(); + } + }, this.rateTime, this.rateTime, this.timeUnit); + } + + synchronized void renew() { + acquiredPermits = 0; + notifyAll(); + } + + @Override + public String toString() { + return Objects.toStringHelper(this).add("rateTime", rateTime).add("permits", permits) + .add("acquiredPermits", acquiredPermits).toString(); + } + +} \ No newline at end of file diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java new file mode 100644 index 0000000000000..a72012a76c997 --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java @@ -0,0 +1,165 @@ +/** + * 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.util; + +import static org.testng.Assert.fail; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertEquals; + +import java.util.concurrent.TimeUnit; + +import org.testng.annotations.Test; + +public class RateLimiterTest { + + @Test + public void testInvalidRenewTime() { + try { + new RateLimiter(0, 100, TimeUnit.SECONDS); + fail("should have thrown exception: invalid rate, must be > 0"); + } catch (IllegalArgumentException ie) { + // Ok + } + + try { + new RateLimiter(10, 0, TimeUnit.SECONDS); + fail("should have thrown exception: invalid rateTime, must be > 0"); + } catch (IllegalArgumentException ie) { + // Ok + } + } + + @Test + public void testShutDown() throws Exception { + RateLimiter rate = new RateLimiter(1, 1000, TimeUnit.MILLISECONDS); + assertFalse(rate.isShutdown()); + rate.shutdown(); + assertTrue(rate.isShutdown()); + try { + rate.acquire(); + fail("should have failed, executor is already closed"); + } catch (IllegalArgumentException e) { + // ok + } + } + + @Test + public void testAcquireBlock() throws Exception { + final long rateTimeMSec = 1000; + RateLimiter rate = new RateLimiter(1, rateTimeMSec, TimeUnit.MILLISECONDS); + rate.acquire(); + assertTrue(rate.getAvailablePermits() == 0); + long start = System.currentTimeMillis(); + rate.acquire(); + long end = System.currentTimeMillis(); + // no permits are available: need to wait on acquire + assertTrue((end - start) > rateTimeMSec / 2); + rate.shutdown(); + } + + @Test + public void testAcquire() throws Exception { + final long rateTimeMSec = 1000; + final int permits = 100; + RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS); + long start = System.currentTimeMillis(); + for (int i = 0; i < permits; i++) { + rate.acquire(); + } + long end = System.currentTimeMillis(); + assertTrue((end - start) < rateTimeMSec); + assertTrue(rate.getAvailablePermits() == 0); + rate.shutdown(); + } + + @Test + public void testMultipleAcquire() throws Exception { + final long rateTimeMSec = 1000; + final int permits = 100; + final int acquirePermist = 50; + RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS); + long start = System.currentTimeMillis(); + for (int i = 0; i < permits / acquirePermist; i++) { + rate.acquire(acquirePermist); + } + long end = System.currentTimeMillis(); + assertTrue((end - start) < rateTimeMSec); + assertTrue(rate.getAvailablePermits() == 0); + rate.shutdown(); + } + + @Test + public void testTryAcquireNoPermits() throws Exception { + final long rateTimeMSec = 1000; + RateLimiter rate = new RateLimiter(1, rateTimeMSec, TimeUnit.MILLISECONDS); + assertTrue(rate.tryAcquire()); + assertFalse(rate.tryAcquire()); + assertTrue(rate.getAvailablePermits() == 0); + rate.shutdown(); + } + + @Test + public void testTryAcquire() throws Exception { + final long rateTimeMSec = 1000; + final int permits = 100; + RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS); + for (int i = 0; i < permits; i++) { + rate.tryAcquire(); + } + assertTrue(rate.getAvailablePermits() == 0); + rate.shutdown(); + } + + @Test + public void testMultipleTryAcquire() throws Exception { + final long rateTimeMSec = 1000; + final int permits = 100; + final int acquirePermist = 50; + RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS); + for (int i = 0; i < permits / acquirePermist; i++) { + rate.tryAcquire(acquirePermist); + } + assertTrue(rate.getAvailablePermits() == 0); + rate.shutdown(); + } + + @Test + public void testResetRate() throws Exception { + final long rateTimeMSec = 1000; + final int permits = 100; + RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS); + rate.tryAcquire(permits); + assertEquals(rate.getAvailablePermits(), 0); + // check after a rate-time: permits must be renewed + Thread.sleep(rateTimeMSec * 2); + assertEquals(rate.getAvailablePermits(), permits); + + // change rate-time from 1sec to 5sec + rate.setRate(permits, 5 * rateTimeMSec, TimeUnit.MILLISECONDS); + rate.tryAcquire(permits); + assertEquals(rate.getAvailablePermits(), 0); + // check after a rate-time: permits can't be renewed + Thread.sleep(rateTimeMSec); + assertEquals(rate.getAvailablePermits(), 0); + + rate.shutdown(); + } + +} diff --git a/site/docs/latest/admin-api/namespaces.md b/site/docs/latest/admin-api/namespaces.md index 924e3174e8426..4327bac80e708 100644 --- a/site/docs/latest/admin-api/namespaces.md +++ b/site/docs/latest/admin-api/namespaces.md @@ -583,6 +583,62 @@ GET /admin/namespaces/{property}/{cluster}/{namespace}/retention admin.namespaces().getRetention(namespace) ``` +#### set dispatch throttling + +It sets message dispatch rate for all the topics under a given namespace. +Dispatch rate can be restricted by number of message per X seconds (`msg-dispatch-rate`) or by number of message-bytes per X second (`byte-dispatch-rate`). +dispatch rate is in second and it can be configured with `dispatch-rate-period`. Default value of `msg-dispatch-rate` and `byte-dispatch-rate` is -1 which +disables the throttling. + +###### CLI + +``` +$ pulsar-admin namespaces set-dispatch-rate test-property/cl1/ns1 --msg-dispatch-rate 1000 --byte-dispatch-rate 1048576 --dispatch-rate-period 1 +``` + +###### REST + +``` +POST /admin/namespaces/{property}/{cluster}/{namespace}/dispatchRate +``` + +###### Java + +```java +admin.namespaces().setDispatchRate(namespace, 1000, 1048576, 1) +``` + +#### get configured message-rate + +It shows configured message-rate for the namespace (topics under this namespace can dispatch this many messages per second) + +###### CLI + +``` +$ pulsar-admin namespaces get-dispatch-rate test-property/cl1/ns1 +``` + +```json +{ + "dispatchThrottlingRatePerTopicInMsg" : 1000, + "dispatchThrottlingRatePerTopicInByte" : 1048576, + "ratePeriodInSecond" : 1 +} +``` + +###### REST + +``` +GET /admin/namespaces/{property}/{cluster}/{namespace}/dispatchRate +``` + +###### Java + +```java +admin.namespaces().getDispatchRate(namespace) +``` + + ### Namespace isolation Coming soon. From 408ebc05958f0cbdd749bab0fd0d3c737b6e5848 Mon Sep 17 00:00:00 2001 From: rdhabalia Date: Fri, 11 Aug 2017 18:00:25 -0700 Subject: [PATCH 2/2] address comment --- conf/broker.conf | 4 ++-- conf/standalone.conf | 4 ++-- .../pulsar/broker/admin/Namespaces.java | 5 ----- .../persistent/DispatchRateLimiter.java | 12 +++++----- .../pulsar/common/util/RateLimiter.java | 19 ++++++++-------- .../pulsar/common/util/RateLimiterTest.java | 22 +++++++++---------- 6 files changed, 31 insertions(+), 35 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 5e2ff92f0af4b..486d8f1561619 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -106,11 +106,11 @@ maxUnackedMessagesPerBroker=0 # limit/2 messages maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16 -# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default +# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default # message dispatch-throttling dispatchThrottlingRatePerTopicInMsg=0 -# Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling +# Default bytes per second dispatch throttling-limit for every topic. Using a value of 0, is disabling # default message-byte dispatch-throttling dispatchThrottlingRatePerTopicInByte=0 diff --git a/conf/standalone.conf b/conf/standalone.conf index a121bc2126901..f1067f6172c83 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -95,11 +95,11 @@ maxUnackedMessagesPerBroker=0 # limit/2 messages maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16 -# Default number of message dispatching throttling-limit for every topic. Using a value of 0, is disabling default +# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default # message dispatch-throttling dispatchThrottlingRatePerTopicInMsg=0 -# Default number of message-bytes dispatching throttling-limit for every topic. Using a value of 0, is disabling +# Default bytes per second dispatch throttling-limit for every topic. Using a value of 0, is disabling # default message-byte dispatch-throttling dispatchThrottlingRatePerTopicInByte=0 diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java index be674fc1415ce..bb333c38ccb8a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java @@ -822,11 +822,6 @@ public void setDispatchRate(@PathParam("property") String property, @PathParam(" dispatchRate); validateSuperUserAccess(); - if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) { - validateClusterOwnership(cluster); - validateClusterForProperty(property, cluster); - } - Entry policiesNode = null; NamespaceName nsName = new NamespaceName(property, cluster, namespace); 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 e2d37a52a5cae..53ea070eff6b9 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 @@ -150,7 +150,7 @@ public DispatchRate getPoliciesDispatchRate() { } /** - * Update dispatch rate by updating msg and byte rate-limiter. If dispatch-rate is configured < 0 then it shutdowns + * 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 @@ -175,7 +175,7 @@ public synchronized void updateDispatchRate(DispatchRate dispatchRate) { } else { // message-rate should be disable and close if (this.dispatchRateLimiterOnMessage != null) { - this.dispatchRateLimiterOnMessage.shutdown(); + this.dispatchRateLimiterOnMessage.close(); this.dispatchRateLimiterOnMessage = null; } } @@ -192,7 +192,7 @@ public synchronized void updateDispatchRate(DispatchRate dispatchRate) { } else { // message-rate should be disable and close if (this.dispatchRateLimiterOnByte != null) { - this.dispatchRateLimiterOnByte.shutdown(); + this.dispatchRateLimiterOnByte.close(); this.dispatchRateLimiterOnByte = null; } } @@ -217,13 +217,13 @@ public long getDispatchRateOnByte() { } public void close() { - // shutdown rate-limiter + // close rate-limiter if (dispatchRateLimiterOnMessage != null) { - dispatchRateLimiterOnMessage.shutdown(); + dispatchRateLimiterOnMessage.close(); dispatchRateLimiterOnMessage = null; } if (dispatchRateLimiterOnByte != null) { - dispatchRateLimiterOnByte.shutdown(); + dispatchRateLimiterOnByte.close(); dispatchRateLimiterOnByte = null; } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java index 25082dd62bc94..5ee602b79e777 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java @@ -53,7 +53,7 @@ * * */ -public class RateLimiter { +public class RateLimiter implements AutoCloseable{ private final ScheduledExecutorService executorService; private long rateTime; @@ -62,7 +62,7 @@ public class RateLimiter { private ScheduledFuture renewTask; private long permits; private long acquiredPermits; - private boolean isShutdown; + private boolean isClosed; public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit) { this(null, permits, rateTime, timeUnit); @@ -90,20 +90,21 @@ public RateLimiter(final ScheduledExecutorService service, final long permits, f } - public synchronized void shutdown() { - if (!isShutdown) { + @Override + public synchronized void close() { + if (!isClosed) { if (!externalExecutor) { executorService.shutdownNow(); } if (renewTask != null) { renewTask.cancel(false); } - isShutdown = true; + isClosed = true; } } - public synchronized boolean isShutdown() { - return isShutdown; + public synchronized boolean isClosed() { + return isClosed; } /** @@ -125,7 +126,7 @@ public synchronized void acquire() throws InterruptedException { * the number of permits to acquire */ public synchronized void acquire(long acquirePermit) throws InterruptedException { - checkArgument(!isShutdown(), "Rate limiter is already shutdown"); + checkArgument(!isClosed(), "Rate limiter is already shutdown"); checkArgument(acquirePermit <= this.permits, "acquiring permits must be less or equal than initialized rate =" + this.permits); @@ -167,7 +168,7 @@ public synchronized boolean tryAcquire() { * @return {@code true} if the permits were acquired, {@code false} otherwise */ public synchronized boolean tryAcquire(long acquirePermit) { - checkArgument(!isShutdown(), "Rate limiter is already shutdown"); + checkArgument(!isClosed(), "Rate limiter is already shutdown"); // lazy init and start task only once application start using it if (renewTask == null) { renewTask = createTask(); diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java index a72012a76c997..dc36d793b5250 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java @@ -47,11 +47,11 @@ public void testInvalidRenewTime() { } @Test - public void testShutDown() throws Exception { + public void testclose() throws Exception { RateLimiter rate = new RateLimiter(1, 1000, TimeUnit.MILLISECONDS); - assertFalse(rate.isShutdown()); - rate.shutdown(); - assertTrue(rate.isShutdown()); + assertFalse(rate.isClosed()); + rate.close(); + assertTrue(rate.isClosed()); try { rate.acquire(); fail("should have failed, executor is already closed"); @@ -71,7 +71,7 @@ public void testAcquireBlock() throws Exception { long end = System.currentTimeMillis(); // no permits are available: need to wait on acquire assertTrue((end - start) > rateTimeMSec / 2); - rate.shutdown(); + rate.close(); } @Test @@ -86,7 +86,7 @@ public void testAcquire() throws Exception { long end = System.currentTimeMillis(); assertTrue((end - start) < rateTimeMSec); assertTrue(rate.getAvailablePermits() == 0); - rate.shutdown(); + rate.close(); } @Test @@ -102,7 +102,7 @@ public void testMultipleAcquire() throws Exception { long end = System.currentTimeMillis(); assertTrue((end - start) < rateTimeMSec); assertTrue(rate.getAvailablePermits() == 0); - rate.shutdown(); + rate.close(); } @Test @@ -112,7 +112,7 @@ public void testTryAcquireNoPermits() throws Exception { assertTrue(rate.tryAcquire()); assertFalse(rate.tryAcquire()); assertTrue(rate.getAvailablePermits() == 0); - rate.shutdown(); + rate.close(); } @Test @@ -124,7 +124,7 @@ public void testTryAcquire() throws Exception { rate.tryAcquire(); } assertTrue(rate.getAvailablePermits() == 0); - rate.shutdown(); + rate.close(); } @Test @@ -137,7 +137,7 @@ public void testMultipleTryAcquire() throws Exception { rate.tryAcquire(acquirePermist); } assertTrue(rate.getAvailablePermits() == 0); - rate.shutdown(); + rate.close(); } @Test @@ -159,7 +159,7 @@ public void testResetRate() throws Exception { Thread.sleep(rateTimeMSec); assertEquals(rate.getAvailablePermits(), 0); - rate.shutdown(); + rate.close(); } }