This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 8cb6af6ae41 [fix][broker] Fixed reset for AggregatedNamespaceStats
(#21225)
8cb6af6ae41 is described below
commit 8cb6af6ae41c3387c767d1dcf67533a5b1c166a7
Author: Asaf Mesika <[email protected]>
AuthorDate: Sun Sep 24 06:43:40 2023 +0300
[fix][broker] Fixed reset for AggregatedNamespaceStats (#21225)
---
pulsar-broker/pom.xml | 1 +
.../stats/prometheus/AggregatedNamespaceStats.java | 24 ++++-
.../prometheus/AggregatedNamespaceStatsTest.java | 103 ++++++++++++++++++++-
3 files changed, 124 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index cb12cdfedfe..8149c02e5ac 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -568,6 +568,7 @@
<artifactId>protobuf-maven-plugin</artifactId>
<version>${protobuf-maven-plugin.version}</version>
<configuration>
+ <!--suppress UnresolvedMavenProperty -->
<protocArtifact>com.google.protobuf:protoc:${protoc3.version}:exe:${os.detected.classifier}</protocArtifact>
<checkStaleness>true</checkStaleness>
<excludes>
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
index 9fe5588044d..d0dc4fe2a7e 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
@@ -34,7 +34,7 @@ public class AggregatedNamespaceStats {
public double throughputIn;
public double throughputOut;
- public long messageAckRate;
+ public double messageAckRate;
public long bytesInCounter;
public long msgInCounter;
public long bytesOutCounter;
@@ -64,7 +64,7 @@ public class AggregatedNamespaceStats {
long compactionCompactedEntriesCount;
long compactionCompactedEntriesSize;
StatsBuckets compactionLatencyBuckets = new
StatsBuckets(CompactionRecord.WRITE_LATENCY_BUCKETS_USEC);
- int delayedMessageIndexSizeInBytes;
+ long delayedMessageIndexSizeInBytes;
Map<String, TopicMetricBean> bucketDelayedIndexStats = new HashMap<>();
@@ -182,14 +182,34 @@ public class AggregatedNamespaceStats {
rateOut = 0;
throughputIn = 0;
throughputOut = 0;
+ messageAckRate = 0;
+ bytesInCounter = 0;
+ msgInCounter = 0;
+
+ bytesOutCounter = 0;
+ msgOutCounter = 0;
msgBacklog = 0;
msgDelayed = 0;
+ ongoingTxnCount = 0;
+ abortedTxnCount = 0;
+ committedTxnCount = 0;
+
backlogQuotaLimit = 0;
backlogQuotaLimitTime = -1;
replicationStats.clear();
subscriptionStats.clear();
+
+ compactionRemovedEventCount = 0;
+ compactionSucceedCount = 0;
+ compactionFailedCount = 0;
+ compactionDurationTimeInMills = 0;
+ compactionReadThroughput = 0;
+ compactionWriteThroughput = 0;
+ compactionCompactedEntriesCount = 0;
+ compactionCompactedEntriesSize = 0;
+
delayedMessageIndexSizeInBytes = 0;
bucketDelayedIndexStats.clear();
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
index b5933f9ecf5..0e12d75f74f 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
@@ -20,10 +20,12 @@ package org.apache.pulsar.broker.stats.prometheus;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-
+import java.util.HashMap;
+import org.apache.bookkeeper.mledger.util.StatsBuckets;
+import org.apache.pulsar.common.policies.data.stats.TopicMetricBean;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = {"broker"})
public class AggregatedNamespaceStatsTest {
@Test
@@ -157,4 +159,101 @@ public class AggregatedNamespaceStatsTest {
assertEquals(nsSubStats.unackedMessages, 2);
}
+
+ @Test
+ public void testReset() {
+ AggregatedNamespaceStats stats = new AggregatedNamespaceStats();
+ stats.topicsCount = 8;
+ stats.subscriptionsCount = 3;
+ stats.producersCount = 1;
+ stats.consumersCount = 8;
+ stats.rateIn = 1.3;
+ stats.rateOut = 3.5;
+ stats.throughputIn = 3.2;
+ stats.throughputOut = 5.8;
+ stats.messageAckRate = 12;
+ stats.bytesInCounter = 1234;
+ stats.msgInCounter = 3889;
+ stats.bytesOutCounter = 89775;
+ stats.msgOutCounter = 28983;
+ stats.msgBacklog = 39;
+ stats.msgDelayed = 31;
+
+ stats.ongoingTxnCount = 87;
+ stats.abortedTxnCount = 74;
+ stats.committedTxnCount = 34;
+
+ stats.backlogQuotaLimit = 387;
+ stats.backlogQuotaLimitTime = 8771;
+
+ stats.replicationStats = new HashMap<>();
+ stats.replicationStats.put("r", new AggregatedReplicationStats());
+
+ stats.subscriptionStats = new HashMap<>();
+ stats.subscriptionStats.put("r", new AggregatedSubscriptionStats());
+
+ stats.compactionRemovedEventCount = 124;
+ stats.compactionSucceedCount = 487;
+ stats.compactionFailedCount = 84857;
+ stats.compactionDurationTimeInMills = 2384;
+ stats.compactionReadThroughput = 355423;
+ stats.compactionWriteThroughput = 23299;
+ stats.compactionCompactedEntriesCount = 37522;
+ stats.compactionCompactedEntriesSize = 8475;
+
+ stats.compactionLatencyBuckets = new StatsBuckets(5);
+ stats.compactionLatencyBuckets.addValue(3);
+
+ stats.delayedMessageIndexSizeInBytes = 45223;
+
+ stats.bucketDelayedIndexStats = new HashMap<>();
+ stats.bucketDelayedIndexStats.put("t", new TopicMetricBean());
+
+ stats.reset();
+
+ assertEquals(stats.bytesOutCounter, 0);
+ assertEquals(stats.topicsCount, 0);
+ assertEquals(stats.subscriptionsCount, 0);
+ assertEquals(stats.producersCount, 0);
+ assertEquals(stats.consumersCount, 0);
+ assertEquals(stats.rateIn, 0);
+ assertEquals(stats.rateOut, 0);
+ assertEquals(stats.throughputIn, 0);
+ assertEquals(stats.throughputOut, 0);
+ assertEquals(stats.messageAckRate, 0);
+ assertEquals(stats.bytesInCounter, 0);
+ assertEquals(stats.msgInCounter, 0);
+ assertEquals(stats.bytesOutCounter, 0);
+ assertEquals(stats.msgOutCounter, 0);
+
+ assertEquals(stats.managedLedgerStats.storageSize, 0);
+
+ assertEquals(stats.msgBacklog, 0);
+ assertEquals(stats.msgDelayed, 0);
+
+ assertEquals(stats.ongoingTxnCount, 0);
+ assertEquals(stats.abortedTxnCount, 0);
+ assertEquals(stats.committedTxnCount, 0);
+
+ assertEquals(stats.backlogQuotaLimit, 0);
+ assertEquals(stats.backlogQuotaLimitTime, -1);
+
+ assertEquals(stats.replicationStats.size(), 0);
+ assertEquals(stats.subscriptionStats.size(), 0);
+
+ assertEquals(stats.compactionRemovedEventCount, 0);
+ assertEquals(stats.compactionSucceedCount, 0);
+ assertEquals(stats.compactionFailedCount, 0);
+ assertEquals(stats.compactionDurationTimeInMills, 0);
+ assertEquals(stats.compactionReadThroughput, 0);
+ assertEquals(stats.compactionWriteThroughput, 0);
+ assertEquals(stats.compactionCompactedEntriesCount, 0);
+ assertEquals(stats.compactionCompactedEntriesSize, 0);
+
+ assertEquals(stats.compactionLatencyBuckets.getSum(), 0);
+
+ assertEquals(stats.delayedMessageIndexSizeInBytes, 0);
+ assertEquals(stats.bucketDelayedIndexStats.size(), 0);
+ }
+
}
