Fix subscription consumer count

(cherry picked from commit 1d4f9a6)
datastax · Apr 6, 2022 · 81927aa · 81927aa
1 parent 5855f99
commit 81927aa
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 1 deletion.
diff --git a/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java b/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
@@ -60,5 +60,7 @@ public class AggregatedSubscriptionStats {
 
     double msgDropRate;
 
+    long consumersCount;
+
     public Map<Consumer, AggregatedConsumerStats> consumerStat = new HashMap<>();
 }
diff --git a/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -32,6 +32,7 @@
 import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
 import org.apache.pulsar.common.policies.data.stats.NonPersistentTopicStatsImpl;
 import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl;
+import org.apache.pulsar.common.policies.data.stats.SubscriptionStatsImpl;
 import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
 import org.apache.pulsar.compaction.CompactedTopicContext;
@@ -189,6 +190,7 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
                 subsStats.lastConsumedFlowTimestamp = subscriptionStats.lastConsumedFlowTimestamp;
                 subsStats.lastConsumedTimestamp = subscriptionStats.lastConsumedTimestamp;
                 subsStats.lastMarkDeleteAdvancedTimestamp = subscriptionStats.lastMarkDeleteAdvancedTimestamp;
+                subsStats.consumersCount = subscriptionStats.consumers.size();
                 subscriptionStats.consumers.forEach(cStats -> {
                     stats.consumersCount++;
                     subsStats.unackedMessages += cStats.unackedMessages;
@@ -222,6 +224,7 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
                 subsStats.lastConsumedFlowTimestamp = subscriptionStats.getLastConsumedFlowTimestamp();
                 subsStats.lastConsumedTimestamp = subscriptionStats.getLastConsumedTimestamp();
                 subsStats.lastMarkDeleteAdvancedTimestamp = subscriptionStats.getLastMarkDeleteAdvancedTimestamp();
+                subsStats.consumersCount = subscriptionStats.getConsumers().size();
                 subsStats.msgDropRate = subscriptionStats.getMsgDropRate();
                 subscriptionStats.getConsumers().forEach(cStats -> {
                     stats.consumersCount++;

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
@@ -265,7 +265,7 @@ static void printTopicStats(SimpleTextOutputStream stream, String cluster, Strin
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_drop_rate",
                     subsStats.msgDropRate, splitTopicAndPartitionIndexLabel);
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_consumers_count",
-                    subsStats.consumerStat.size(), splitTopicAndPartitionIndexLabel);
+                    subsStats.consumersCount, splitTopicAndPartitionIndexLabel);
             subsStats.consumerStat.forEach((c, consumerStats) -> {
                 metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                         "pulsar_consumer_msg_rate_redeliver", consumerStats.msgRateRedeliver,