[pulsar] branch branch-2.10 updated: [fix][broker] Fix issue where msgRateExpired may not refresh forever (#19759)

[technoboy]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.10 by this push:
     new 9a9f0054fc3 [fix][broker] Fix issue where msgRateExpired may not 
refresh forever (#19759)
9a9f0054fc3 is described below

commit 9a9f0054fc344b37e71c6932ce4edbf345df589a
Author: Masahiro Sakamoto <massa...@yahoo-corp.jp>
AuthorDate: Sun Mar 12 14:39:30 2023 +0900

    [fix][broker] Fix issue where msgRateExpired may not refresh forever 
(#19759)
---
 .../service/persistent/PersistentReplicator.java   |  2 ++
 .../broker/service/persistent/PersistentTopic.java |  1 +
 .../client/api/SimpleProducerConsumerStatTest.java | 41 ++++++++++++++++++++++
 3 files changed, 44 insertions(+)

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
index 86a46a5c9c2..76a032c0217 100644
--- 
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
@@ -682,6 +682,8 @@ public class PersistentReplicator extends AbstractReplicator
     public void updateRates() {
         msgOut.calculateRate();
         msgExpired.calculateRate();
+        expiryMonitor.updateRates();
+
         stats.msgRateOut = msgOut.getRate();
         stats.msgThroughputOut = msgOut.getValueRate();
         stats.msgRateExpired = msgExpired.getRate() + 
expiryMonitor.getMessageExpiryRate();
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
index fde4cb7d07f..1d8ba95b2f7 100644
--- 
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
@@ -1797,6 +1797,7 @@ public class PersistentTopic extends AbstractTopic 
implements Topic, AddEntryCal
                 // 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);
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
index 40d34a7c0b7..0a609631da1 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
@@ -453,4 +453,45 @@ public class SimpleProducerConsumerStatTest extends 
ProducerConsumerBase {
                 .until(() -> producer.getStats().getPendingQueueSize() == 
numMessages);
         assertEquals(producer.getStats().getPendingQueueSize(), numMessages);
     }
+
+    @Test
+    public void testMsgRateExpired() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        String topicName = "persistent://my-property/tp1/my-ns/" + methodName;
+        String subName = "my-sub";
+        admin.topics().createSubscription(topicName, subName, 
MessageId.latest);
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topicName)
+                .enableBatching(false)
+                .create();
+
+        int numMessages = 100;
+        for (int i = 0; i < numMessages; i++) {
+            String message = "my-message-" + i;
+            producer.send(message.getBytes());
+        }
+
+        Thread.sleep(2000);
+        admin.topics().expireMessages(topicName, subName, 1);
+        pulsar.getBrokerService().updateRates();
+
+        Awaitility.await().ignoreExceptions().timeout(5, TimeUnit.SECONDS)
+                .until(() -> 
admin.topics().getStats(topicName).getSubscriptions().get(subName).getMsgRateExpired()
 > 0.001);
+
+        Thread.sleep(2000);
+        pulsar.getBrokerService().updateRates();
+
+        Awaitility.await().ignoreExceptions().timeout(5, TimeUnit.SECONDS)
+                .until(() -> 
admin.topics().getStats(topicName).getSubscriptions().get(subName).getMsgRateExpired()
 < 0.001);
+
+        
assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getMsgRateExpired(),
 0.0,
+                0.001);
+        
assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getTotalMsgExpired(),
+                numMessages);
+
+        log.info("-- Exiting {} test --", methodName);
+    }
 }