From d120541743d412c4280088017add8401bab239e1 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Fri, 7 Feb 2025 17:21:06 +0800 Subject: [PATCH 1/6] [improve][pip] PIP-406: Introduce pulsar_subscription_dispatch_throttled_msgs and bytes metrics --- pip/pip-406.md | 75 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) create mode 100644 pip/pip-406.md diff --git a/pip/pip-406.md b/pip/pip-406.md new file mode 100644 index 0000000000000..4db42e80337bd --- /dev/null +++ b/pip/pip-406.md @@ -0,0 +1,75 @@ +# PIP-406: Introduce pulsar_subscription_dispatch_throttled_msgs and bytes metrics + +# Background knowledge + +## Motivation + +Currently, users can monitor subscription backlogs using the `pulsar_subscription_back_log_no_delayed` metric. +However, if [dispatch throttling](https://pulsar.apache.org/docs/next/concepts-throttling/) is configured at the broker/topic/subscription level, +this metric may not accurately reflect whether the backlog is due to insufficient consumer capacity, as it could be caused by dispatch throttling. + +## Goals + +Introduce metrics to indicate the number of `messages/bytes throttled` for a subscription. This allows users to write PromQL queries to identify subscriptions with high backlogs but low or no throttling, pinpointing backlogs caused by insufficient consumer capacity. + +## In Scope +- Introduce the metric `pulsar_subscription_dispatch_throttled_msgs` to represent the total number of messages throttled for a subscription. +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes` to represent the total number of bytes throttled for a subscription. +- Add `dispatchThrottledMsgs` and `dispatchThrottledBytes` fields to topic subscription stats. + +## Out of Scope +- These states are not persistent and will reset upon subscription reconnection. + +# High Level Design +1. Maintain `dispatchThrottledMsgs` and `dispatchThrottledBytes` in `AbstractBaseDispatcher`. Increase these values whenever the number of messages/bytes is reduced during `calculateToRead`. +2. Output these fields when retrieving topic stats and metrics. + + +# Detailed Design + +## Design & Implementation Details +1. Maintain `dispatchThrottledMsgs` and `dispatchThrottledBytes` in `AbstractBaseDispatcher`: +```java + private final LongAdder dispatchThrottledMsgs = new LongAdder(); + private final AtomicLong dispatchThrottledBytes = new AtomicLong(); +``` + +2. During each [calculateToRead](https://github.com/apache/pulsar/blob/411f6973e85b0a6213e992386e1704f93d0aae42/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L371-L377), +if the number of `messages/bytes` is reduced, increase these fields accordingly. + +- dispatchThrottledBytes may overflow in extreme cases, so reset this value before overflow: +```diff + protected Pair updateMessagesToRead(DispatchRateLimiter dispatchRateLimiter, + int messagesToRead, long bytesToRead) { + // update messagesToRead according to available dispatch rate limit. +- return computeReadLimits(messagesToRead, ++ Pair result = computeReadLimits(messagesToRead, + (int) dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg(), + bytesToRead, dispatchRateLimiter.getAvailableDispatchRateLimitOnByte()); ++ if (result.getLeft() < messagesToRead) { ++ dispatchThrottledMsgs.add(messagesToRead - result.getLeft()); ++ } ++ if (result.getRight() < bytesToRead) { ++ long increment = bytesToRead - result.getRight(); ++ dispatchThrottledBytes.updateAndGet(current -> { ++ // Check if adding the increment would cause an overflow ++ if (Long.MAX_VALUE - current < increment) { ++ return increment; ++ } ++ return current + increment; ++ }); ++ } ++ return result; + } +``` + +## Public-facing Changes +- None + + +### Configuration +- None + +# Backward & Forward Compatibility +- Full Compatibility + From 6f8bd911ae546cec6b1080204c2aa0daea175b19 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Mon, 17 Feb 2025 09:19:37 +0800 Subject: [PATCH 2/6] update public facing --- pip/pip-406.md | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/pip/pip-406.md b/pip/pip-406.md index 4db42e80337bd..eb8f58b3dae09 100644 --- a/pip/pip-406.md +++ b/pip/pip-406.md @@ -64,7 +64,29 @@ if the number of `messages/bytes` is reduced, increase these fields accordingly. ``` ## Public-facing Changes -- None + +### Metrics + +- Introduce the metric `pulsar_subscription_dispatch_throttled_msgs` to represent the total number of messages throttled for a subscription. +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes` to represent the total number of bytes throttled for a subscription. + +1. pulsar_subscription_dispatch_throttled_msgs: + - Description: The total number of messages throttled for a subscription. + - Attributes: + - tenant + - namespace + - topic + - subscription + - Unit: messages count + +2. pulsar_subscription_dispatch_throttled_bytes: + - Description: The total number of bytes throttled for a subscription. + - Attributes: + - tenant + - namespace + - topic + - subscription + - Unit: messages bytes ### Configuration From e7d9e6cd5885e55628c547f9d1581b73f86e532d Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Tue, 18 Feb 2025 17:02:20 +0800 Subject: [PATCH 3/6] new design --- pip/pip-406.md | 155 +++++++++++++++++++++++++++++++++---------------- 1 file changed, 105 insertions(+), 50 deletions(-) diff --git a/pip/pip-406.md b/pip/pip-406.md index eb8f58b3dae09..fbea157d7210e 100644 --- a/pip/pip-406.md +++ b/pip/pip-406.md @@ -1,4 +1,4 @@ -# PIP-406: Introduce pulsar_subscription_dispatch_throttled_msgs and bytes metrics +# PIP-406: Introduce metrics related to dispatch_throttled_count # Background knowledge @@ -10,84 +10,139 @@ this metric may not accurately reflect whether the backlog is due to insufficien ## Goals -Introduce metrics to indicate the number of `messages/bytes throttled` for a subscription. This allows users to write PromQL queries to identify subscriptions with high backlogs but low or no throttling, pinpointing backlogs caused by insufficient consumer capacity. +Introduce metrics to indicate the count of `messages/bytes throttled` for **broker/topic/subscription** level rate limit. +This allows users to write PromQL queries to identify subscriptions with high backlogs but low or no throttling, pinpointing backlogs caused by insufficient consumer capacity. ## In Scope -- Introduce the metric `pulsar_subscription_dispatch_throttled_msgs` to represent the total number of messages throttled for a subscription. -- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes` to represent the total number of bytes throttled for a subscription. -- Add `dispatchThrottledMsgs` and `dispatchThrottledBytes` fields to topic subscription stats. + +Broker Level: +- Introduce the metric `pulsar_broker_dispatch_throttled_msg_count` to represent the total count of messages throttled for a broker. +- Introduce the metric `pulsar_broker_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a broker. + +Topic Level: +- Introduce the metric `pulsar_dispatch_throttled_msg_count` to represent the total count of messages throttled for a topic. +- Introduce the metric `pulsar_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a topic. + +Subscription Level: +- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_count` to represent the total count of messages throttled for a subscription. +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a subscription. + ## Out of Scope -- These states are not persistent and will reset upon subscription reconnection. +- These states are not persistent and will reset upon broker restart/ topic re-load / subscription reconnected. # High Level Design -1. Maintain `dispatchThrottledMsgs` and `dispatchThrottledBytes` in `AbstractBaseDispatcher`. Increase these values whenever the number of messages/bytes is reduced during `calculateToRead`. -2. Output these fields when retrieving topic stats and metrics. +1. Maintain `dispatchThrottleMsgCount` and `dispatchThrottleBytesCount` in `DispatchRateLimiter`. Increase these values in the `consumeDispatchQuota` method when the TokenBucket for messages or bytes is insufficient. +2. Output these fields when retrieving metrics. # Detailed Design ## Design & Implementation Details -1. Maintain `dispatchThrottledMsgs` and `dispatchThrottledBytes` in `AbstractBaseDispatcher`: +1. Maintain `dispatchThrottleMsgCount` and `dispatchThrottleBytesCount` in `DispatchRateLimiter`: ```java - private final LongAdder dispatchThrottledMsgs = new LongAdder(); - private final AtomicLong dispatchThrottledBytes = new AtomicLong(); + private final LongAdder dispatchThrottleMsgCount = new LongAdder(); + private final LongAdder dispatchThrottleBytesCount = new LongAdder(); ``` -2. During each [calculateToRead](https://github.com/apache/pulsar/blob/411f6973e85b0a6213e992386e1704f93d0aae42/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L371-L377), -if the number of `messages/bytes` is reduced, increase these fields accordingly. - -- dispatchThrottledBytes may overflow in extreme cases, so reset this value before overflow: +2. During each [consumeDispatchQuota](https://github.com/apache/pulsar/blob/c4cff0ab3dac169c0a1418ef2f63f61604f6278e/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java#L97-L104), +if token bucket is insufficient, increase these fields accordingly. ```diff - protected Pair updateMessagesToRead(DispatchRateLimiter dispatchRateLimiter, - int messagesToRead, long bytesToRead) { - // update messagesToRead according to available dispatch rate limit. -- return computeReadLimits(messagesToRead, -+ Pair result = computeReadLimits(messagesToRead, - (int) dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg(), - bytesToRead, dispatchRateLimiter.getAvailableDispatchRateLimitOnByte()); -+ if (result.getLeft() < messagesToRead) { -+ dispatchThrottledMsgs.add(messagesToRead - result.getLeft()); -+ } -+ if (result.getRight() < bytesToRead) { -+ long increment = bytesToRead - result.getRight(); -+ dispatchThrottledBytes.updateAndGet(current -> { -+ // Check if adding the increment would cause an overflow -+ if (Long.MAX_VALUE - current < increment) { -+ return increment; -+ } -+ return current + increment; -+ }); -+ } -+ return result; + public void consumeDispatchQuota(long numberOfMessages, long byteSize) { + AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage; + if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) { +- localDispatchRateLimiterOnMessage.consumeTokens(numberOfMessages); ++ if (!localDispatchRateLimiterOnMessage.consumeTokensAndCheckIfContainsTokens(numberOfMessages)) { ++ dispatchThrottleMsgCount.increment(); ++ } + } + AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte; + if (byteSize > 0 && localDispatchRateLimiterOnByte != null) { +- localDispatchRateLimiterOnByte.consumeTokens(byteSize); ++ if (!localDispatchRateLimiterOnByte.consumeTokensAndCheckIfContainsTokens(byteSize)) { ++ dispatchThrottleBytesCount.increment(); ++ } + } } ``` +3. When get stats or metrics for broker/topic/subscription throttle count, output these fields. ## Public-facing Changes ### Metrics -- Introduce the metric `pulsar_subscription_dispatch_throttled_msgs` to represent the total number of messages throttled for a subscription. -- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes` to represent the total number of bytes throttled for a subscription. - -1. pulsar_subscription_dispatch_throttled_msgs: - - Description: The total number of messages throttled for a subscription. +1. `pulsar_broker_dispatch_throttled_msg_count`: + - Description: The total count of messages throttled for a broker - Attributes: + - cluster + - Unit: throttled count + +2. `pulsar_broker_dispatch_throttled_bytes_count`: + - Description: The total count of bytes throttled for a broker + - Attributes: + - cluster + - Unit: throttled count + +3. `pulsar_dispatch_throttled_msg_count`: + - Description: The total count of messages throttled for a topic + - Attributes: + - tenant + - namespace + - topic + - Unit: messages count + +4. `pulsar_dispatch_throttled_bytes_count`: + - Description: The total count of messages throttled for a topic + - Attributes: + - tenant + - namespace + - topic + - Unit: messages count + +5. `pulsar_subscription_dispatch_throttled_msg_count`: + - Description: The total count of messages throttled for a subscription + - Attributes: - tenant - namespace - topic - subscription - Unit: messages count - -2. pulsar_subscription_dispatch_throttled_bytes: - - Description: The total number of bytes throttled for a subscription. - - Attributes: - - tenant - - namespace - - topic - - subscription - - Unit: messages bytes +6. `pulsar_subscription_dispatch_throttled_bytes_count`: + - Description: The total count of messages throttled for a subscription + - Attributes: + - tenant + - namespace + - topic + - subscription + - Unit: messages count + +### API +1. Add get throttle count interface on admin api `TopicStats`. +```java + /** + * Total count of throttling occurrences due to message rate limiting. + */ + long getDispatchThrottledMsgCount(); + + /** + * Total count of throttling occurrences due to byte rate limiting. + */ + long getDispatchThrottledBytesCount(); +``` + +2. Add get throttle count interface on admin api `SubscriptionStats`. +```java + /** + * Total count of throttling occurrences due to message rate limiting. + */ + long getDispatchThrottledMsgCount(); + + /** + * Total count of throttling occurrences due to byte rate limiting. + */ + long getDispatchThrottledBytesCount(); +``` ### Configuration - None From 2494112ae04c850a0d90ff9a48c9d583762deca8 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Fri, 28 Feb 2025 15:21:15 +0800 Subject: [PATCH 4/6] Address code reviews --- pip/pip-406.md | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/pip/pip-406.md b/pip/pip-406.md index fbea157d7210e..b89d6c25719a8 100644 --- a/pip/pip-406.md +++ b/pip/pip-406.md @@ -45,25 +45,26 @@ Subscription Level: private final LongAdder dispatchThrottleBytesCount = new LongAdder(); ``` -2. During each [consumeDispatchQuota](https://github.com/apache/pulsar/blob/c4cff0ab3dac169c0a1418ef2f63f61604f6278e/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java#L97-L104), -if token bucket is insufficient, increase these fields accordingly. +2. Each time a read occurs, if the expected number of messages or bytes is reduced, increment the metric by one. ```diff - public void consumeDispatchQuota(long numberOfMessages, long byteSize) { - AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage; - if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) { -- localDispatchRateLimiterOnMessage.consumeTokens(numberOfMessages); -+ if (!localDispatchRateLimiterOnMessage.consumeTokensAndCheckIfContainsTokens(numberOfMessages)) { -+ dispatchThrottleMsgCount.increment(); -+ } + private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimiter, + MutablePair readLimits, + DispatchRateLimiter.Type limiterType) { ++ int originalMessagesToRead = readLimits.getLeft(); ++ long originalBytesToRead = readLimits.getRight(); + // update messagesToRead according to available dispatch rate limit. + int availablePermitsOnMsg = (int) rateLimiter.getAvailableDispatchRateLimitOnMsg(); + if (availablePermitsOnMsg >= 0) { +@@ -414,6 +416,12 @@ private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimi + if (availablePermitsOnByte >= 0) { + readLimits.setRight(Math.min(readLimits.getRight(), availablePermitsOnByte)); } - AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte; - if (byteSize > 0 && localDispatchRateLimiterOnByte != null) { -- localDispatchRateLimiterOnByte.consumeTokens(byteSize); -+ if (!localDispatchRateLimiterOnByte.consumeTokensAndCheckIfContainsTokens(byteSize)) { -+ dispatchThrottleBytesCount.increment(); -+ } - } - } ++ if (readLimits.getLeft() < originalMessagesToRead) { ++ rateLimiter.increaseDispatchThrottleMsgCount(); ++ } ++ if (readLimits.getRight() < originalBytesToRead) { ++ rateLimiter.increaseDispatchThrottleBytesCount(); ++ } ``` 3. When get stats or metrics for broker/topic/subscription throttle count, output these fields. From 577910c58b4bb156c049b29b4af3fef437e07572 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Mon, 3 Mar 2025 17:27:31 +0800 Subject: [PATCH 5/6] Change docs --- pip/pip-406.md | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/pip/pip-406.md b/pip/pip-406.md index b89d6c25719a8..7140287bccdaa 100644 --- a/pip/pip-406.md +++ b/pip/pip-406.md @@ -1,4 +1,4 @@ -# PIP-406: Introduce metrics related to dispatch_throttled_count +# PIP-406: Introduce metrics related to dispatch throttled number of times # Background knowledge @@ -16,16 +16,16 @@ This allows users to write PromQL queries to identify subscriptions with high ba ## In Scope Broker Level: -- Introduce the metric `pulsar_broker_dispatch_throttled_msg_count` to represent the total count of messages throttled for a broker. -- Introduce the metric `pulsar_broker_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a broker. +- Introduce the metric `pulsar_broker_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a broker due to broker rate limits. +- Introduce the metric `pulsar_broker_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a broker due to broker rate limits. Topic Level: -- Introduce the metric `pulsar_dispatch_throttled_msg_count` to represent the total count of messages throttled for a topic. -- Introduce the metric `pulsar_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a topic. +- Introduce the metric `pulsar_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a topic due to topic rate limits. +- Introduce the metric `pulsar_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a topic due to topic rate limits. Subscription Level: -- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_count` to represent the total count of messages throttled for a subscription. -- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_count` to represent the total count of bytes throttled for a subscription. +- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a subscription due to subscription rate limits. +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a subscription due to subscription rate limits. ## Out of Scope @@ -73,19 +73,19 @@ Subscription Level: ### Metrics 1. `pulsar_broker_dispatch_throttled_msg_count`: - - Description: The total count of messages throttled for a broker + - Description: The total number of times message dispatching was throttled on a broker due to broker rate limits - Attributes: - cluster - Unit: throttled count 2. `pulsar_broker_dispatch_throttled_bytes_count`: - - Description: The total count of bytes throttled for a broker + - Description: The total number of times byte dispatching was throttled on a broker due to broker rate limits - Attributes: - cluster - Unit: throttled count 3. `pulsar_dispatch_throttled_msg_count`: - - Description: The total count of messages throttled for a topic + - Description: The total number of times message dispatching was throttled on a topic due to topic rate limits - Attributes: - tenant - namespace @@ -93,7 +93,7 @@ Subscription Level: - Unit: messages count 4. `pulsar_dispatch_throttled_bytes_count`: - - Description: The total count of messages throttled for a topic + - Description: The total number of times byte dispatching was throttled on a topic due to topic rate limits - Attributes: - tenant - namespace @@ -101,7 +101,7 @@ Subscription Level: - Unit: messages count 5. `pulsar_subscription_dispatch_throttled_msg_count`: - - Description: The total count of messages throttled for a subscription + - Description: The total number of times message dispatching was throttled on a subscription due to subscription rate limits - Attributes: - tenant - namespace @@ -110,7 +110,7 @@ Subscription Level: - Unit: messages count 6. `pulsar_subscription_dispatch_throttled_bytes_count`: - - Description: The total count of messages throttled for a subscription + - Description: The total number of times byte dispatching was throttled on a subscription due to subscription rate limits - Attributes: - tenant - namespace From 699067c271a62f7a63e0dbc7c324c811ce4a24b8 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 13 Mar 2025 19:58:36 +0800 Subject: [PATCH 6/6] Update design --- pip/pip-406.md | 195 +++++++++++++++++++++++++++---------------------- 1 file changed, 109 insertions(+), 86 deletions(-) diff --git a/pip/pip-406.md b/pip/pip-406.md index 7140287bccdaa..b541d48c380ce 100644 --- a/pip/pip-406.md +++ b/pip/pip-406.md @@ -1,4 +1,4 @@ -# PIP-406: Introduce metrics related to dispatch throttled number of times +# PIP-406: Introduce metrics related to dispatch throttled events # Background knowledge @@ -15,34 +15,31 @@ This allows users to write PromQL queries to identify subscriptions with high ba ## In Scope -Broker Level: -- Introduce the metric `pulsar_broker_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a broker due to broker rate limits. -- Introduce the metric `pulsar_broker_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a broker due to broker rate limits. +- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_events` to represent the total number of times message dispatching was throttled on a subscription +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_events` to represent the total number of times byte dispatching was throttled on a subscription -Topic Level: -- Introduce the metric `pulsar_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a topic due to topic rate limits. -- Introduce the metric `pulsar_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a topic due to topic rate limits. - -Subscription Level: -- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_count` to represent the total number of times message dispatching was throttled on a subscription due to subscription rate limits. -- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_count` to represent the total number of times byte dispatching was throttled on a subscription due to subscription rate limits. - +Since throttling on a subscription can be caused by multiple rate limit(broker/topic/subscription), +these metrics will include a label `reason: broker/topic/subscription` to indicate which throttler is responsible. ## Out of Scope - These states are not persistent and will reset upon broker restart/ topic re-load / subscription reconnected. # High Level Design -1. Maintain `dispatchThrottleMsgCount` and `dispatchThrottleBytesCount` in `DispatchRateLimiter`. Increase these values in the `consumeDispatchQuota` method when the TokenBucket for messages or bytes is insufficient. +1. Maintain metrics counters for throttling caused by `broker/topic/subscription` rate limit. Increment the appropriate counter whenever throttling occurs. 2. Output these fields when retrieving metrics. # Detailed Design ## Design & Implementation Details -1. Maintain `dispatchThrottleMsgCount` and `dispatchThrottleBytesCount` in `DispatchRateLimiter`: +1. Maintain these fields in AbstractBaseDispatcher. ```java - private final LongAdder dispatchThrottleMsgCount = new LongAdder(); - private final LongAdder dispatchThrottleBytesCount = new LongAdder(); + public final LongAdder dispatchThrottledMsgEventsBySubscriptionLimit = new LongAdder(); + public final LongAdder dispatchThrottledMsgEventsByTopicLimit = new LongAdder(); + public final LongAdder dispatchThrottledMsgEventsByBrokerLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsBySubscriptionLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsByTopicLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsByBrokerLimit = new LongAdder(); ``` 2. Each time a read occurs, if the expected number of messages or bytes is reduced, increment the metric by one. @@ -50,99 +47,125 @@ Subscription Level: private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimiter, MutablePair readLimits, DispatchRateLimiter.Type limiterType) { -+ int originalMessagesToRead = readLimits.getLeft(); -+ long originalBytesToRead = readLimits.getRight(); - // update messagesToRead according to available dispatch rate limit. - int availablePermitsOnMsg = (int) rateLimiter.getAvailableDispatchRateLimitOnMsg(); - if (availablePermitsOnMsg >= 0) { -@@ -414,6 +416,12 @@ private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimi - if (availablePermitsOnByte >= 0) { - readLimits.setRight(Math.min(readLimits.getRight(), availablePermitsOnByte)); - } -+ if (readLimits.getLeft() < originalMessagesToRead) { -+ rateLimiter.increaseDispatchThrottleMsgCount(); -+ } -+ if (readLimits.getRight() < originalBytesToRead) { -+ rateLimiter.increaseDispatchThrottleBytesCount(); -+ } ++ int originalMessagesToRead = readLimits.getLeft(); ++ long originalBytesToRead = readLimits.getRight(); + int availablePermitsOnMsg = (int) rateLimiter.getAvailableDispatchRateLimitOnMsg(); + if (availablePermitsOnMsg >= 0) { + readLimits.setLeft(Math.min(readLimits.getLeft(), availablePermitsOnMsg)); + } + long availablePermitsOnByte = rateLimiter.getAvailableDispatchRateLimitOnByte(); + if (availablePermitsOnByte >= 0) { + readLimits.setRight(Math.min(readLimits.getRight(), availablePermitsOnByte)); + } ++ if (readLimits.getLeft() < originalMessagesToRead) { ++ switch (limiterType) { ++ case BROKER -> dispatchThrottledMsgEventsByBrokerLimit.increment(); ++ case TOPIC -> dispatchThrottledMsgEventsByTopicLimit.increment(); ++ case SUBSCRIPTION -> dispatchThrottledMsgEventsBySubscriptionLimit.increment(); ++ default -> {} ++ } ++ } ++ if (readLimits.getRight() < originalBytesToRead) { ++ switch (limiterType) { ++ case BROKER -> dispatchThrottledBytesEventsByBrokerLimit.increment(); ++ case TOPIC -> dispatchThrottledBytesEventsByTopicLimit.increment(); ++ case SUBSCRIPTION -> dispatchThrottledBytesEventsBySubscriptionLimit.increment(); ++ default -> {} ++ } ++ } +``` +3. Print these metrics when retrieving metrics(TopicStats#printTopicStats). +```java +// write dispatch throttling metrics with `reason` labels to identify specific throttling +// causes: by subscription limit, by topic limit, or by broker limit. +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsBySubscriptionLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "subscription"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsBySubscriptionLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "subscription"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsByTopicLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "topic"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsByTopicLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "topic"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsByBrokerLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "broker"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsByBrokerLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "broker");} ``` -3. When get stats or metrics for broker/topic/subscription throttle count, output these fields. ## Public-facing Changes ### Metrics -1. `pulsar_broker_dispatch_throttled_msg_count`: - - Description: The total number of times message dispatching was throttled on a broker due to broker rate limits - - Attributes: - - cluster - - Unit: throttled count - -2. `pulsar_broker_dispatch_throttled_bytes_count`: - - Description: The total number of times byte dispatching was throttled on a broker due to broker rate limits - - Attributes: - - cluster - - Unit: throttled count - -3. `pulsar_dispatch_throttled_msg_count`: - - Description: The total number of times message dispatching was throttled on a topic due to topic rate limits - - Attributes: - - tenant - - namespace - - topic - - Unit: messages count - -4. `pulsar_dispatch_throttled_bytes_count`: - - Description: The total number of times byte dispatching was throttled on a topic due to topic rate limits - - Attributes: - - tenant - - namespace - - topic - - Unit: messages count - -5. `pulsar_subscription_dispatch_throttled_msg_count`: - - Description: The total number of times message dispatching was throttled on a subscription due to subscription rate limits +1. `pulsar_subscription_dispatch_throttled_msg_events`: + - Description: The total number of times message dispatching was throttled on a subscription - Attributes: - tenant - namespace - topic - subscription + - `reason`: broker/topic/subscription - Unit: messages count -6. `pulsar_subscription_dispatch_throttled_bytes_count`: - - Description: The total number of times byte dispatching was throttled on a subscription due to subscription rate limits +2. `pulsar_subscription_dispatch_throttled_bytes_events`: + - Description: The total number of times byte dispatching was throttled on a subscription - Attributes: - tenant - namespace - topic - subscription + - `reason`: broker/topic/subscription - Unit: messages count ### API -1. Add get throttle count interface on admin api `TopicStats`. +Add get throttle count interface on admin api `SubscriptionStats`. ```java /** - * Total count of throttling occurrences due to message rate limiting. - */ - long getDispatchThrottledMsgCount(); - - /** - * Total count of throttling occurrences due to byte rate limiting. - */ - long getDispatchThrottledBytesCount(); -``` - -2. Add get throttle count interface on admin api `SubscriptionStats`. -```java - /** - * Total count of throttling occurrences due to message rate limiting. - */ - long getDispatchThrottledMsgCount(); - - /** - * Total count of throttling occurrences due to byte rate limiting. - */ - long getDispatchThrottledBytesCount(); + * Gets the total number of times message dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled message events by subscription limit, default is 0. + */ +long getDispatchThrottledMsgEventsBySubscriptionLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled bytes by subscription limit, default is 0. + */ +long getDispatchThrottledBytesEventsBySubscriptionLimit(); + +/** + * Gets the total number of times message dispatching was throttled on a subscription due to topic rate limits. + * @return the count of throttled message events by topic limit, default is 0. + */ +long getDispatchThrottledMsgEventsByTopicLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to topic rate limits. + * @return the count of throttled bytes events by topic limit, default is 0. + */ +long getDispatchThrottledBytesEventsByTopicLimit(); + +/** + * Gets the total number of times message dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled message events by broker limit, default is 0. + */ +long getDispatchThrottledMsgEventsByBrokerLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled bytes events by broker limit, default is 0. + */ +long getDispatchThrottledBytesEventsByBrokerLimit(); ``` ### Configuration