PIP-3 : Introduce message-dispatch rate limiting

conf/broker.conf

@@ -106,6 +106,14 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages   
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16
 
+# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default
+# message dispatch-throttling 
+dispatchThrottlingRatePerTopicInMsg=0
+
+# Default bytes per second dispatch 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
 

conf/standalone.conf

@@ -95,6 +95,14 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages  
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16
 
+# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default
+# message dispatch-throttling 
+dispatchThrottlingRatePerTopicInMsg=0
+
+# Default bytes per second dispatch 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
 

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;
     }

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<Policies> policiesCache() {
         return pulsar().getConfigurationCache().policiesCache();
     }
 
+    ZooKeeperDataCache<LocalPolicies> localPoliciesCache() {
+        return pulsar().getLocalZkCacheService().policiesCache();
+    }
+
     ZooKeeperDataCache<ClusterData> clustersCache() {
         return pulsar().getConfigurationCache().clustersCache();
     }

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,61 @@ 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();
+
+        Entry<Policies, Stat> 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.")

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<Optional<LocalPolicies>> createPolicies(String path, boolean readFromGlobal) {
-        checkNotNull(path, "path can't be null");
-        checkArgument(path.startsWith(LOCAL_POLICIES_ROOT), "Invalid path of local policies");
-
         CompletableFuture<Optional<LocalPolicies>> 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);

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;

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<Policies> {
@@ -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.

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<ChannelPromise, Integer> sendMessages(final List<Entry> entries) {
+    public SendMessageInfo sendMessages(final List<Entry> entries) {
         final ChannelHandlerContext ctx = cnx.ctx();
-        final MutablePair<ChannelPromise, Integer> sentMessages = new MutablePair<ChannelPromise, Integer>();
+        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<Entry> entries) throws PulsarServerException {
+    void updatePermitsAndPendingAcks(final List<Entry> entries, SendMessageInfo sentMessages) throws PulsarServerException {
         int permitsToReduce = 0;
         Iterator<Entry> iter = entries.iterator();
         boolean unsupportedVersion = false;
@@ -276,7 +278,8 @@ int updatePermitsAndPendingAcks(final List<Entry> 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);
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java

@@ -145,7 +145,7 @@ public SubType getType() {
     public void sendMessages(List<Entry> 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);