[improve][broker] Add msgInReplay subscription stat and metric to improve Key_Shared observability

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java

@@ -149,4 +149,13 @@ public boolean containsStickyKeyHashes(Set<Integer> stickyKeyHashes) {
     public NavigableSet<Position> getMessagesToReplayNow(int maxMessagesToRead) {
         return messagesToRedeliver.items(maxMessagesToRead, PositionFactory::create);
     }
+
+    /**
+     * Get the number of messages registered for replay in the redelivery controller.
+     *
+     * @return number of messages
+     */
+    public int size() {
+        return messagesToRedeliver.size();
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -135,7 +135,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
     protected final ExecutorService dispatchMessagesThread;
     private final SharedConsumerAssignor assignor;
 
-
     protected enum ReadType {
         Normal, Replay
     }
@@ -1352,5 +1351,9 @@ public Subscription getSubscription() {
         return subscription;
     }
 
+    public long getNumberOfMessagesInReplay() {
+        return redeliveryMessages.size();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(PersistentDispatcherMultipleConsumers.class);
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -1276,6 +1276,7 @@ public CompletableFuture<SubscriptionStatsImpl> getStatsAsync(GetStatsOptions ge
                 subStats.unackedMessages = d.getTotalUnackedMessages();
                 subStats.blockedSubscriptionOnUnackedMsgs = d.isBlockedDispatcherOnUnackedMsgs();
                 subStats.msgDelayed = d.getNumberOfDelayedMessages();
+                subStats.msgInReplay = d.getNumberOfMessagesInReplay();
             }
         }
         subStats.msgBacklog = getNumberOfEntriesInBacklog(getStatsOptions.isGetPreciseBacklog());

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java

@@ -43,6 +43,7 @@ public class AggregatedNamespaceStats {
     public ManagedLedgerStats managedLedgerStats = new ManagedLedgerStats();
     public long msgBacklog;
     public long msgDelayed;
+    public long msgInReplay;
 
     public long ongoingTxnCount;
     public long abortedTxnCount;
@@ -141,10 +142,12 @@ void updateStats(TopicStats stats) {
             AggregatedSubscriptionStats subsStats =
                     subscriptionStats.computeIfAbsent(n, k -> new AggregatedSubscriptionStats());
             msgDelayed += as.msgDelayed;
+            msgInReplay += as.msgInReplay;
             subsStats.blockedSubscriptionOnUnackedMsgs = as.blockedSubscriptionOnUnackedMsgs;
             subsStats.msgBacklog += as.msgBacklog;
             subsStats.msgBacklogNoDelayed += as.msgBacklogNoDelayed;
             subsStats.msgDelayed += as.msgDelayed;
+            subsStats.msgInReplay += as.msgInReplay;
             subsStats.msgRateRedeliver += as.msgRateRedeliver;
             subsStats.unackedMessages += as.unackedMessages;
             subsStats.filterProcessedMsgCount += as.filterProcessedMsgCount;
@@ -200,6 +203,7 @@ public void reset() {
 
         msgBacklog = 0;
         msgDelayed = 0;
+        msgInReplay = 0;
         ongoingTxnCount = 0;
         abortedTxnCount = 0;
         committedTxnCount = 0;

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java

@@ -43,6 +43,8 @@ public class AggregatedSubscriptionStats {
 
     public long msgDelayed;
 
+    public long msgInReplay;
+
     long msgOutCounter;
 
     long bytesOutCounter;

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java

@@ -134,6 +134,7 @@ private static void aggregateTopicStats(TopicStats stats, SubscriptionStatsImpl
         subsStats.msgOutCounter = subscriptionStats.msgOutCounter;
         subsStats.msgBacklog = subscriptionStats.msgBacklog;
         subsStats.msgDelayed = subscriptionStats.msgDelayed;
+        subsStats.msgInReplay = subscriptionStats.msgInReplay;
         subsStats.msgRateExpired = subscriptionStats.msgRateExpired;
         subsStats.totalMsgExpired = subscriptionStats.totalMsgExpired;
         subsStats.msgBacklogNoDelayed = subsStats.msgBacklog - subsStats.msgDelayed;
@@ -424,6 +425,8 @@ private static void printNamespaceStats(PrometheusMetricStreams stream, Aggregat
 
         writeMetric(stream, "pulsar_subscription_delayed", stats.msgDelayed, cluster, namespace);
 
+        writeMetric(stream, "pulsar_subscription_in_replay", stats.msgInReplay, cluster, namespace);
+
         writeMetric(stream, "pulsar_delayed_message_index_size_bytes", stats.delayedMessageIndexSizeInBytes, cluster,
                 namespace);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java

@@ -310,6 +310,8 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st
                     subsStats.msgBacklogNoDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel);
             writeSubscriptionMetric(stream, "pulsar_subscription_delayed",
                     subsStats.msgDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel);
+            writeSubscriptionMetric(stream, "pulsar_subscription_in_replay",
+                    subsStats.msgInReplay, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel);
             writeSubscriptionMetric(stream, "pulsar_subscription_msg_rate_redeliver",
                     subsStats.msgRateRedeliver, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel);
             writeSubscriptionMetric(stream, "pulsar_subscription_unacked_messages",

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java

@@ -66,6 +66,9 @@ public interface SubscriptionStats {
     /** Number of delayed messages currently being tracked. */
     long getMsgDelayed();
 
+    /** Number of messages registered for replay. */
+    long getMsgInReplay();
+
     /**
      * Number of unacknowledged messages for the subscription, where an unacknowledged message is one that has been
      * sent to a consumer but not yet acknowledged. Calculated by summing all {@link ConsumerStats#getUnackedMessages()}

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java

@@ -74,6 +74,9 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
     /** Number of delayed messages currently being tracked. */
     public long msgDelayed;
 
+    /** Number of messages registered for replay. */
+    public long msgInReplay;
+
     /**
      * Number of unacknowledged messages for the subscription, where an unacknowledged message is one that has been
      * sent to a consumer but not yet acknowledged. Calculated by summing all {@link ConsumerStatsImpl#unackedMessages}
@@ -173,6 +176,8 @@ public void reset() {
         msgBacklog = 0;
         backlogSize = 0;
         msgBacklogNoDelayed = 0;
+        msgDelayed = 0;
+        msgInReplay = 0;
         unackedMessages = 0;
         type = null;
         msgRateExpired = 0;
@@ -208,6 +213,7 @@ public SubscriptionStatsImpl add(SubscriptionStatsImpl stats) {
         this.backlogSize += stats.backlogSize;
         this.msgBacklogNoDelayed += stats.msgBacklogNoDelayed;
         this.msgDelayed += stats.msgDelayed;
+        this.msgInReplay += stats.msgInReplay;
         this.unackedMessages += stats.unackedMessages;
         this.type = stats.type;
         this.msgRateExpired += stats.msgRateExpired;