You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/07/22 10:02:42 UTC
(pulsar) 07/15: [fix][broker] Fix bucket delayed message index metrics reset on scrape (#26171)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 819b2cc7da44464c502f56a2665df041f9cae36a
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Jul 10 10:41:18 2026 +0800
[fix][broker] Fix bucket delayed message index metrics reset on scrape (#26171)
(cherry picked from commit c240dd2ea2b8365dff921190c5af8ba7380914f6)
---
.../bookkeeper/mledger/util/StatsBuckets.java | 28 +++++++++
.../bucket/BucketDelayedMessageIndexStats.java | 8 +--
.../bucket/BucketDelayedMessageIndexStatsTest.java | 68 ++++++++++++++++++++++
3 files changed, 100 insertions(+), 4 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/StatsBuckets.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/StatsBuckets.java
index 60c0a7f6c9d2..ec3a7e00fd39 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/StatsBuckets.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/StatsBuckets.java
@@ -60,6 +60,16 @@ public class StatsBuckets {
sumCounter.add(value);
}
+ /**
+ * Snapshots the current values into {@link #getBuckets()}, {@link #getCount()}, and
+ * {@link #getSum()}, then resets the internal counters to zero.
+ *
+ * <p>Use this for periodic aggregation where the consumer reads values on a fixed
+ * schedule (e.g. {@code ManagedLedgerMBeanImpl} refreshes every
+ * {@code managedLedgerStatsPeriodSeconds}). Do <b>not</b> use this for metrics that
+ * are scraped on demand (e.g. Prometheus {@code /metrics}), because data recorded
+ * between scrapes is lost after each call — use {@link #snapshot()} instead.
+ */
public void refresh() {
long count = 0;
sum = sumCounter.sumThenReset();
@@ -73,6 +83,24 @@ public class StatsBuckets {
this.count = count;
}
+ /**
+ * Snapshots the current cumulative values without resetting the internal counters.
+ * Use this instead of {@link #refresh()} when the data should survive across reads
+ * (e.g. Prometheus metrics that are scraped multiple times).
+ */
+ public void snapshot() {
+ long count = 0;
+ sum = sumCounter.sum();
+
+ for (int i = 0; i < buckets.length; i++) {
+ long value = buckets[i].sum();
+ count += value;
+ values[i] = value;
+ }
+
+ this.count = count;
+ }
+
public void reset() {
sum = 0;
sumCounter.reset();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java
index 68788c359d56..b9e6c7dc64c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java
@@ -47,8 +47,8 @@ public class BucketDelayedMessageIndexStats {
private static final String BUCKET_TOTAL_NAME = "pulsar_delayed_message_index_bucket_total";
private static final String INDEX_LOADED_NAME = "pulsar_delayed_message_index_loaded";
private static final String SNAPSHOT_SIZE_BYTES_NAME = "pulsar_delayed_message_index_bucket_snapshot_size_bytes";
- private static final String OP_COUNT_NAME = "pulsar_delayed_message_index_bucket_op_count";
- private static final String OP_LATENCY_NAME = "pulsar_delayed_message_index_bucket_op_latency_ms";
+ static final String OP_COUNT_NAME = "pulsar_delayed_message_index_bucket_op_count";
+ static final String OP_LATENCY_NAME = "pulsar_delayed_message_index_bucket_op_latency_ms";
private final AtomicInteger delayedMessageIndexBucketTotal = new AtomicInteger();
private final AtomicLong delayedMessageIndexLoaded = new AtomicLong();
@@ -75,11 +75,11 @@ public class BucketDelayedMessageIndexStats {
String[] labels = splitKey(k);
String[] labelsAndValues = new String[] {"state", labels[0], "type", labels[1]};
String key = OP_COUNT_NAME + joinKey(labelsAndValues);
- metrics.put(key, new TopicMetricBean(OP_COUNT_NAME, count.sumThenReset(), labelsAndValues));
+ metrics.put(key, new TopicMetricBean(OP_COUNT_NAME, count.sum(), labelsAndValues));
});
delayedMessageIndexBucketOpLatencyMs.forEach((typeName, statsBuckets) -> {
- statsBuckets.refresh();
+ statsBuckets.snapshot();
long[] buckets = statsBuckets.getBuckets();
for (int i = 0; i < buckets.length; i++) {
long count = buckets[i];
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStatsTest.java
new file mode 100644
index 000000000000..b1b7d8248cc3
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStatsTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.delayed.bucket;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import java.util.Map;
+import org.apache.pulsar.common.policies.data.stats.TopicMetricBean;
+import org.testng.annotations.Test;
+
+public class BucketDelayedMessageIndexStatsTest {
+
+ private static final String OP_COUNT = BucketDelayedMessageIndexStats.OP_COUNT_NAME;
+ private static final String OP_LATENCY_COUNT = BucketDelayedMessageIndexStats.OP_LATENCY_NAME + "_count";
+ private static final String OP_LATENCY_SUM = BucketDelayedMessageIndexStats.OP_LATENCY_NAME + "_sum";
+
+ @Test
+ public void testMetricsAreCumulativeAcrossScrapes() {
+ BucketDelayedMessageIndexStats stats = new BucketDelayedMessageIndexStats();
+ var type = BucketDelayedMessageIndexStats.Type.create;
+
+ stats.recordTriggerEvent(type);
+ stats.recordSuccessEvent(type, 200);
+ stats.recordSuccessEvent(type, 300);
+
+ // First scrape
+ var m1 = stats.genTopicMetricMap();
+ assertEquals(get(m1, OP_COUNT, "state", "succeed", "type", "create"), 2);
+ assertEquals(get(m1, OP_LATENCY_COUNT, "type", "create"), 2);
+ assertEquals(get(m1, OP_LATENCY_SUM, "type", "create"), 500);
+
+ // Second scrape — must not reset
+ var m2 = stats.genTopicMetricMap();
+ assertEquals(get(m2, OP_COUNT, "state", "succeed", "type", "create"), 2);
+ assertEquals(get(m2, OP_LATENCY_COUNT, "type", "create"), 2);
+ assertEquals(get(m2, OP_LATENCY_SUM, "type", "create"), 500);
+
+ // Third event — cumulative
+ stats.recordSuccessEvent(type, 100);
+ var m3 = stats.genTopicMetricMap();
+ assertEquals(get(m3, OP_COUNT, "state", "succeed", "type", "create"), 3);
+ assertEquals(get(m3, OP_LATENCY_COUNT, "type", "create"), 3);
+ assertEquals(get(m3, OP_LATENCY_SUM, "type", "create"), 600);
+ }
+
+ private static long get(Map<String, TopicMetricBean> metrics, String name, String... labelsAndValues) {
+ String key = name + BucketDelayedMessageIndexStats.joinKey(labelsAndValues);
+ TopicMetricBean bean = metrics.get(key);
+ assertNotNull(bean, "Metric not found: " + key);
+ return (long) bean.value;
+ }
+}