[improve][broker]Add the expired message count metric from this subscription within the last interval. (#24106)

Co-authored-by: zjxxzjwang <[email protected]>

Author: zjxxzjwang

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

@@ -173,6 +173,10 @@ public double getMessageExpiryRate() {
         return msgExpired.getRate();
     }
 
+    public long getMessageExpiryCount() {
+        return msgExpired.getCount();
+    }
+
     public long getTotalMessageExpired() {
         return totalMsgExpired.sum();
     }

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

@@ -1348,6 +1348,7 @@ public CompletableFuture<SubscriptionStatsImpl> getStatsAsync(GetStatsOptions ge
         }
         subStats.msgBacklogNoDelayed = subStats.msgBacklog - subStats.msgDelayed;
         subStats.msgRateExpired = expiryMonitor.getMessageExpiryRate();
+        subStats.msgExpired = expiryMonitor.getMessageExpiryCount();
         subStats.totalMsgExpired = expiryMonitor.getTotalMessageExpired();
         subStats.isReplicated = isReplicated();
         subStats.subscriptionProperties = subscriptionProperties;

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

@@ -93,6 +93,9 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
     /** Total rate of messages expired on this subscription (msg/s). */
     public double msgRateExpired;
 
+    /** The count of expired messages on this subscription. */
+    public long msgExpired;
+
     /** Total messages expired on this subscription. */
     public long totalMsgExpired;
 
@@ -209,6 +212,7 @@ public void reset() {
         unackedMessages = 0;
         type = null;
         msgRateExpired = 0;
+        msgExpired = 0;
         totalMsgExpired = 0;
         lastExpireTimestamp = 0L;
         lastMarkDeleteAdvancedTimestamp = 0L;
@@ -254,6 +258,7 @@ public SubscriptionStatsImpl add(SubscriptionStatsImpl stats) {
         this.unackedMessages += stats.unackedMessages;
         this.type = stats.type;
         this.msgRateExpired += stats.msgRateExpired;
+        this.msgExpired += stats.msgExpired;
         this.totalMsgExpired += stats.totalMsgExpired;
         this.isReplicated |= stats.isReplicated;
         this.isDurable |= stats.isDurable;