[improve][broker] Optimize message expiration rate repeated update issues (apache#24073)

zjxxzjwang · zjxxzjwang · web-flow · commit b936f46d2dd2 · 2025-03-19T11:15:18.000-07:00
Co-authored-by: zjxxzjwang &lt;zjxxzjwang@tencent.com&gt;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -170,11 +170,12 @@ public boolean expireMessages(Position messagePosition) {
     }
 
 
-    public void updateRates() {
+    private void updateRates() {
         msgExpired.calculateRate();
     }
 
     public double getMessageExpiryRate() {
+        updateRates();
         return msgExpired.getRate();
     }
 
@@ -190,7 +191,6 @@ public void markDeleteComplete(Object ctx) {
             long numMessagesExpired = (long) ctx - cursor.getNumberOfEntriesInBacklog(false);
             msgExpired.recordMultipleEvents(numMessagesExpired, 0 /* no value stats */);
             totalMsgExpired.add(numMessagesExpired);
-            updateRates();
             // If the subscription is a Key_Shared subscription, we should to trigger message dispatch.
             if (subscription != null && subscription.getType() == SubType.Key_Shared) {
                 subscription.getDispatcher().markDeletePositionMoveForward();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -639,7 +639,6 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
     public void updateRates() {
         msgOut.calculateRate();
         msgExpired.calculateRate();
-        expiryMonitor.updateRates();
 
         stats.msgRateOut = msgOut.getRate();
         stats.msgThroughputOut = msgOut.getValueRate();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2457,7 +2457,6 @@ 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);

Original file line number	Diff line number	Diff line change
`@@ -170,11 +170,12 @@ public boolean expireMessages(Position messagePosition) {`
`170`	`170`	`}`
`171`	`171`
`172`	`172`
`173`		`- public void updateRates() {`
	`173`	`+ private void updateRates() {`
`174`	`174`	`msgExpired.calculateRate();`
`175`	`175`	`}`
`176`	`176`
`177`	`177`	`public double getMessageExpiryRate() {`
	`178`	`+ updateRates();`
`178`	`179`	`return msgExpired.getRate();`
`179`	`180`	`}`
`180`	`181`
`@@ -190,7 +191,6 @@ public void markDeleteComplete(Object ctx) {`
`190`	`191`	`long numMessagesExpired = (long) ctx - cursor.getNumberOfEntriesInBacklog(false);`
`191`	`192`	`msgExpired.recordMultipleEvents(numMessagesExpired, 0 /* no value stats */);`
`192`	`193`	`totalMsgExpired.add(numMessagesExpired);`
`193`		`- updateRates();`
`194`	`194`	`// If the subscription is a Key_Shared subscription, we should to trigger message dispatch.`
`195`	`195`	`if (subscription != null && subscription.getType() == SubType.Key_Shared) {`
`196`	`196`	`subscription.getDispatcher().markDeletePositionMoveForward();`