[fix][broker] Resolve the issue of frequent updates in message expiration deletion rate (#24190)

Co-authored-by: zjxxzjwang <zjxxzjwang@tencent.com>

Author: zjxxzjwang

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

@@ -165,12 +165,11 @@ public boolean expireMessages(Position messagePosition) {
     }
 
 
-    private void updateRates() {
+    public void updateRates() {
         msgExpired.calculateRate();
     }
 
     public double getMessageExpiryRate() {
-        updateRates();
         return msgExpired.getRate();
     }
 

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

@@ -639,6 +639,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
     public void updateRates() {
         msgOut.calculateRate();
         msgExpired.calculateRate();
+        expiryMonitor.updateRates();
 
         stats.msgRateOut = msgOut.getRate();
         stats.msgThroughputOut = msgOut.getValueRate();

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

@@ -2536,6 +2536,7 @@ public void updateRates(NamespaceStats nsStats, NamespaceBundleStats bundleStats
                 // Populate subscription specific stats here
                 topicStatsStream.writePair("msgBacklog",
                         subscription.getNumberOfEntriesInBacklog(true));
+                subscription.getExpiryMonitor().updateRates();
                 topicStatsStream.writePair("msgRateExpired", subscription.getExpiredMessageRate());
                 topicStatsStream.writePair("msgRateOut", subMsgRateOut);
                 topicStatsStream.writePair("messageAckRate", subMsgAckRate);

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java

@@ -810,6 +810,10 @@ public void testPerTopicExpiredStat() throws Exception {
                 pulsar.getBrokerService().getTopicIfExists(topic1).get().get().getSubscription(subName);
         PersistentSubscription sub2 = (PersistentSubscription)
                 pulsar.getBrokerService().getTopicIfExists(topic2).get().get().getSubscription(subName);
+        Awaitility.await().until(() -> sub.getExpiryMonitor().getTotalMessageExpired() != 0);
+        Awaitility.await().until(() -> sub2.getExpiryMonitor().getTotalMessageExpired() != 0);
+        sub.getExpiryMonitor().updateRates();
+        sub2.getExpiryMonitor().updateRates();
         Awaitility.await().until(() -> sub.getExpiredMessageRate() != 0.0);
         Awaitility.await().until(() -> sub2.getExpiredMessageRate() != 0.0);