(pulsar) branch master updated: [fix][broker] Fix inflated rate stats of newly created consumers (#26928)

[pe...@apache.org]

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

codelipenghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 7e7c916854ec [fix][broker] Fix inflated rate stats of newly created consumers (#26928)
7e7c916854ec is described below

commit 7e7c916854ec434c437054774a4a256b58a947fe
Author: Penghui Li <pe...@apache.org>
AuthorDate: Fri Oct 16 09:39:44 2026 -0700

    [fix][broker] Fix inflated rate stats of newly created consumers (#26928)
---
 .../org/apache/pulsar/broker/service/Consumer.java | 14 +++++---
 .../pulsar/broker/stats/ConsumerStatsTest.java     | 41 ++++++++++++++++++++++
 2 files changed, 51 insertions(+), 4 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index ac6b352386b6..a75aaaaa6e22 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -109,6 +109,7 @@ public class Consumer {
     private volatile long firstConsumedFlowTimestamp;
     private volatile long lastConsumedFlowTimestamp;
     private Rate chunkedMessageRate;
+    private long lastRatesUpdateNanos = System.nanoTime();
 
     // Represents how many messages we can safely send to the consumer without
     // overflowing its receiving queue. The consumer will use Flow commands to
@@ -1179,10 +1180,15 @@ public class Consumer {
     }
 
     public void updateRates() {
-        msgOut.calculateRate();
-        chunkedMessageRate.calculateRate();
-        msgRedeliver.calculateRate();
-        messageAckRate.calculateRate();
+        long now = System.nanoTime();
+        // at least 1s, so the burst a new consumer receives while filling its receiver queue
+        // isn't extrapolated into a huge per-second rate
+        double periodInSeconds = Math.max((now - lastRatesUpdateNanos) / 1e9, 1);
+        lastRatesUpdateNanos = now;
+        msgOut.calculateRate(periodInSeconds);
+        chunkedMessageRate.calculateRate(periodInSeconds);
+        msgRedeliver.calculateRate(periodInSeconds);
+        messageAckRate.calculateRate(periodInSeconds);
 
         stats.msgRateOut = msgOut.getRate();
         stats.msgThroughputOut = msgOut.getValueRate();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
index efbe792a90b7..6b1313a9827a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
@@ -113,6 +113,9 @@ public class ConsumerStatsTest extends ProducerConsumerBase {
         // wait for shutdown of the broker, this prevents flakiness which could be caused by metrics being
         // unregistered asynchronously. This impacts the execution of the next test method if this would be happening.
         conf.setBrokerShutdownTimeoutMs(5000L);
+        // only the tests update the rates, so the periodic stats task can't reset the rates they check
+        conf.setStatsUpdateInitialDelayInSecs(100000000);
+        conf.setStatsUpdateFrequencyInSecs(100000000);
         return conf;
     }
 
@@ -435,6 +438,44 @@ public class ConsumerStatsTest extends ProducerConsumerBase {
         }
     }
 
+    @Test
+    public void testNewConsumerRateIsNotInflatedByItsInitialBurst() throws Exception {
+        final String topicName = newUniqueName("persistent://my-property/my-ns/new-consumer-rate");
+        final int messages = 100;
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topicName)
+                .enableBatching(false)
+                .create();
+        for (int i = 0; i < messages; i++) {
+            producer.send(new byte[1024]);
+        }
+        admin.topics().createSubscription(topicName, "sub", MessageId.earliest);
+        // reload the topic, so that it is as young as the consumer
+        admin.topics().unload(topicName);
+
+        // like the compaction reader, a new consumer receives a burst as soon as it sends its flow permits
+        long consumerCreationLowerBound = System.nanoTime();
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(topicName)
+                .subscriptionName("sub")
+                .subscribe();
+        org.apache.pulsar.broker.service.Consumer brokerConsumer = pulsar.getBrokerService()
+                .getTopicReference(topicName).orElseThrow().getSubscription("sub").getConsumers().get(0);
+        Awaitility.await().until(() -> brokerConsumer.getMsgOutCounter() == messages);
+        pulsar.getBrokerService().updateRates();
+        double maxPeriodSeconds = Math.max((System.nanoTime() - consumerCreationLowerBound) / 1e9, 1);
+
+        // a per-second rate can't exceed what the consumer received during the period
+        long bytes = brokerConsumer.getBytesOutCounter();
+        ConsumerStatsImpl stats = brokerConsumer.getStats();
+        assertThat(stats.msgRateOut).as("msgRateOut")
+                .isBetween(messages / maxPeriodSeconds, (double) messages);
+        assertThat(stats.msgThroughputOut).as("msgThroughputOut")
+                .isBetween(bytes / maxPeriodSeconds, (double) bytes);
+    }
+
     @Test
     public void testAvgMessagesPerEntry() throws Exception {
         conf.setAllowOverrideEntryFilters(true);