(pulsar) branch branch-2.10 updated: fix duplicate calculation for msgRateIn and msgThroughputIn in replication stats (#15062)

[yubiao]

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

yubiao 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 e6524dcedec fix duplicate calculation for msgRateIn and 
msgThroughputIn in replication stats (#15062)
e6524dcedec is described below

commit e6524dcedecd5667080910c29931055212876434
Author: Qiang Huang <hqeb...@users.noreply.github.com>
AuthorDate: Fri Apr 22 17:06:27 2022 +0800

    fix duplicate calculation for msgRateIn and msgThroughputIn in replication 
stats (#15062)
    
    (cherry picked from commit 3e1176a6a047d0949d241ae906d80f9302450a66)
---
 .../broker/stats/prometheus/NamespaceStatsAggregator.java   | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
index 253f134b7c2..9ab783c4160 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -251,15 +251,18 @@ public class NamespaceStatsAggregator {
         }
 
         topic.getReplicators().forEach((cluster, replicator) -> {
-            AggregatedReplicationStats aggReplStats = 
stats.replicationStats.computeIfAbsent(cluster,
-                    k -> new AggregatedReplicationStats());
-
             ReplicatorStatsImpl replStats = replicator.getStats();
+            AggregatedReplicationStats aggReplStats = 
stats.replicationStats.get(replicator.getRemoteCluster());
+            if (aggReplStats == null) {
+                aggReplStats = new AggregatedReplicationStats();
+                stats.replicationStats.put(replicator.getRemoteCluster(), 
aggReplStats);
+                aggReplStats.msgRateIn = replStats.msgRateIn;
+                aggReplStats.msgThroughputIn = replStats.msgThroughputIn;
+            }
+
             aggReplStats.msgRateOut += replStats.msgRateOut;
             aggReplStats.msgThroughputOut += replStats.msgThroughputOut;
             aggReplStats.replicationBacklog += replStats.replicationBacklog;
-            aggReplStats.msgRateIn += replStats.msgRateIn;
-            aggReplStats.msgThroughputIn += replStats.msgThroughputIn;
             aggReplStats.msgRateExpired += replStats.msgRateExpired;
             aggReplStats.connectedCount += replStats.connected ? 1 : 0;
             aggReplStats.replicationDelayInSeconds += 
replStats.replicationDelayInSeconds;