You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2026/08/27 02:24:06 UTC
(pulsar) branch branch-4.2 updated: [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric (#26313)
This is an automated email from the ASF dual-hosted git repository.
Technoboy- pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 02c00d56c344 [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric (#26313)
02c00d56c344 is described below
commit 02c00d56c34474a0f8427d07f51cbf05a1b9940b
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Wed Aug 26 18:25:38 2026 +0800
[improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric (#26313)
---
conf/broker.conf | 6 +
conf/standalone.conf | 6 +
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 6 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 9 +
.../pulsar/broker/service/BrokerService.java | 32 +++
.../service/persistent/PersistentSubscription.java | 2 +
.../broker/service/persistent/PersistentTopic.java | 254 ++++++++++++++++++++-
.../prometheus/AggregatedSubscriptionStats.java | 2 +
.../stats/prometheus/NamespaceStatsAggregator.java | 5 +-
.../pulsar/broker/stats/prometheus/TopicStats.java | 6 +
.../broker/service/BacklogQuotaManagerTest.java | 182 ++++++++++++++-
.../prometheus/NamespaceStatsAggregatorTest.java | 7 +-
.../common/policies/data/SubscriptionStats.java | 12 +
.../policies/data/stats/SubscriptionStatsImpl.java | 13 ++
.../data/stats/SubscriptionStatsImplTest.java | 19 +-
15 files changed, 545 insertions(+), 16 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 84ccd1d852b9..bb647d51d43b 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1884,6 +1884,12 @@ exposeBundlesMetricsInPrometheus=false
# Default is false.
exposeCustomTopicMetricLabelsEnabled=false
+# Enable computing the age of the oldest unacknowledged message for each subscription and exposing it
+# through topic stats and Prometheus.
+# When disabled, the broker skips computing per-subscription backlog age and the admin API field
+# SubscriptionStats.oldestBacklogMessageAgeSeconds remains -1. Default is false.
+exposeSubscriptionBacklogAgeInPrometheus=false
+
# A comma-separated list of Topic Property keys that are allowed to be exposed as metrics.
# Only these keys can be set as custom metric labels on topics.
# Example: sla_tier,data_sensitivity,cost_center,app_owner
diff --git a/conf/standalone.conf b/conf/standalone.conf
index c1c3289fa7d1..1c34851a0151 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -1181,6 +1181,12 @@ exposePublisherStats=true
# Default is false.
exposePreciseBacklogInPrometheus=false
+# Enable computing the age of the oldest unacknowledged message for each subscription and exposing it
+# through topic stats and Prometheus.
+# When disabled, the broker skips computing per-subscription backlog age and the admin API field
+# SubscriptionStats.oldestBacklogMessageAgeSeconds remains -1. Default is false.
+exposeSubscriptionBacklogAgeInPrometheus=false
+
# Enable splitting topic and partition label in Prometheus.
# If enabled, a topic name will split into 2 parts, one is topic name without partition index,
# another one is partition index, e.g. (topic=xxx, partition=0).
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index ca326dbba33a..7d13f8b557a1 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -54,6 +54,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
@@ -324,7 +325,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Getter
private final OrderedScheduler scheduledExecutor;
- @Getter
protected final ExecutorService executor;
@Getter
@@ -334,6 +334,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
protected final ManagedLedgerMBeanImpl mbean;
protected final Clock clock;
+ public Executor getExecutor() {
+ return executor;
+ }
+
private static final AtomicLongFieldUpdater<ManagedLedgerImpl> READ_OP_COUNT_UPDATER = AtomicLongFieldUpdater
.newUpdater(ManagedLedgerImpl.class, "readOpCount");
private volatile long readOpCount = 0;
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 4a627a5de495..eb2ccf9d245a 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -3585,6 +3585,15 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private boolean exposeSubscriptionBacklogSizeInPrometheus = false;
+ @FieldContext(
+ category = CATEGORY_METRICS,
+ doc = "Enable computing the age of the oldest unacknowledged message for each subscription and exposing "
+ + "it through topic stats and Prometheus.\n"
+ + " When disabled, the broker skips computing per-subscription backlog age and "
+ + "SubscriptionStats.oldestBacklogMessageAgeSeconds remains -1. Default is false."
+ )
+ private boolean exposeSubscriptionBacklogAgeInPrometheus = false;
+
@FieldContext(
category = CATEGORY_METRICS,
doc = "Enable splitting topic and partition label in Prometheus.\n"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 334dda0f3390..e5b15ce6c936 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -26,6 +26,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.client.util.RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
import static org.apache.pulsar.client.util.RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
import static org.apache.pulsar.common.naming.SystemTopicNames.isTransactionInternalName;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
import com.google.common.util.concurrent.RateLimiter;
@@ -258,6 +259,7 @@ public class BrokerService implements Closeable {
@Getter
private final SingleThreadNonConcurrentFixedRateScheduler backlogQuotaChecker;
+ private final SingleThreadNonConcurrentFixedRateScheduler subscriptionBacklogAgeChecker;
protected final AtomicReference<Semaphore> lookupRequestSemaphore;
@Getter
@@ -395,6 +397,8 @@ public class BrokerService implements Closeable {
new SingleThreadNonConcurrentFixedRateScheduler("pulsar-consumed-ledgers-monitor");
this.backlogQuotaManager = new BacklogQuotaManager(pulsar);
this.backlogQuotaChecker = new SingleThreadNonConcurrentFixedRateScheduler("pulsar-backlog-quota-checker");
+ this.subscriptionBacklogAgeChecker =
+ new SingleThreadNonConcurrentFixedRateScheduler("pulsar-subscription-backlog-age-checker");
this.authenticationService = new AuthenticationService(pulsar.getConfiguration(),
pulsar.getOpenTelemetry().getOpenTelemetry());
this.topicFactory = createPersistentTopicFactory();
@@ -667,6 +671,7 @@ public class BrokerService implements Closeable {
this.startCompactionMonitor();
this.startConsumedLedgersMonitor();
this.startBacklogQuotaChecker();
+ this.startSubscriptionBacklogAgeChecker();
this.updateBrokerPublisherThrottlingMaxRate();
this.updateBrokerDispatchThrottlingMaxRate();
this.startCheckReplicationPolicies();
@@ -803,6 +808,19 @@ public class BrokerService implements Closeable {
}
+ protected void startSubscriptionBacklogAgeChecker() {
+ if (pulsar().getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus()) {
+ final int interval = pulsar().getConfiguration().getBacklogQuotaCheckIntervalInSeconds();
+ log.info("Scheduling a thread to refresh subscription backlog age in background after [{}] seconds",
+ interval);
+ subscriptionBacklogAgeChecker.scheduleAtFixedRateNonConcurrently(
+ catchingAndLoggingThrowables(() -> refreshSubscriptionBacklogAge().join()), interval, interval,
+ TimeUnit.SECONDS);
+ } else {
+ log.info("Subscription backlog age monitoring is disabled");
+ }
+ }
+
public void close() throws IOException {
try {
closeAsync().get();
@@ -956,6 +974,7 @@ public class BrokerService implements Closeable {
compactionMonitor,
consumedLedgersMonitor,
backlogQuotaChecker,
+ subscriptionBacklogAgeChecker,
topicOrderedExecutor,
deduplicationSnapshotMonitor)
.handle());
@@ -2434,6 +2453,19 @@ public class BrokerService implements Closeable {
});
}
+ public CompletableFuture<Void> refreshSubscriptionBacklogAge() {
+ if (!pulsar.getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ forEachPersistentTopic(topic -> futures.add(topic.updateSubscriptionOldPositionInfo()));
+ return FutureUtil.waitForAll(futures).exceptionally(throwable -> {
+ log.warn("Error when refreshSubscriptionBacklogAge()", throwable);
+ return null;
+ });
+ }
+
public CompletableFuture<Boolean> isTopicNsOwnedByBrokerAsync(TopicName topicName) {
return pulsar.getNamespaceService().isServiceUnitOwnedAsync(topicName)
.handle((hasOwnership, t) -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 7438b2eb764b..1c24f689b92f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -1440,6 +1440,8 @@ public class PersistentSubscription extends AbstractSubscription {
}
}
subStats.msgBacklog = getNumberOfEntriesInBacklog(getStatsOptions.isGetPreciseBacklog());
+ subStats.oldestBacklogMessageAgeSeconds =
+ topic.getBestEffortOldestUnacknowledgedMessageAgeSeconds(subName);
if (getStatsOptions.isSubscriptionBacklogSize()) {
subStats.backlogSize = topic.getManagedLedger()
.getEstimatedBacklogSize(cursor.getMarkDeletedPosition());
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 f25a57a44680..978122343d6a 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
@@ -314,6 +314,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
PersistentTopic.class,
OldestPositionInfo.class,
"oldestPositionInfo");
+ private volatile Map<String, OldestPositionInfo> subscriptionOldestPositionInfos;
+
@Value
private static class OldestPositionInfo {
Position oldestCursorMarkDeletePosition;
@@ -3825,6 +3827,44 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
}
+ long getBestEffortOldestUnacknowledgedMessageAgeSeconds(String subscriptionName) {
+ if (!brokerService.pulsar().getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus()) {
+ return -1;
+ }
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos == null) {
+ return -1;
+ }
+ OldestPositionInfo positionInfo = positionInfos.get(subscriptionName);
+ if (positionInfo == null) {
+ return -1;
+ } else {
+ return TimeUnit.MILLISECONDS.toSeconds(
+ Clock.systemUTC().millis() - positionInfo.getPositionPublishTimestampInMillis());
+ }
+ }
+
+ private Map<String, OldestPositionInfo> getSubscriptionOldestPositionInfos() {
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos == null) {
+ synchronized (this) {
+ positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos == null) {
+ positionInfos = new ConcurrentHashMap<>();
+ subscriptionOldestPositionInfos = positionInfos;
+ }
+ }
+ }
+ return positionInfos;
+ }
+
+ private void clearSubscriptionOldestPositionInfos() {
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos != null) {
+ positionInfos.clear();
+ }
+ }
+
private void updateResultIfNewer(OldestPositionInfo updatedResult) {
TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.updateAndGet(this,
existingResult -> {
@@ -3839,6 +3879,212 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ getSubscriptionOldestPositionInfos().compute(subscriptionName,
+ (__, existingResult) -> {
+ if (existingResult == null
+ || ManagedCursorContainer.DataVersion.compareVersions(
+ updatedResult.getDataVersion(), existingResult.getDataVersion()) > 0) {
+ return updatedResult;
+ } else {
+ return existingResult;
+ }
+ });
+ }
+
+ private void clearSubscriptionResultIfNewer(String subscriptionName, long dataVersion) {
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos == null) {
+ return;
+ }
+ positionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ OldestPositionInfo positionInfo = positionInfos == null ? null : positionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Set<String> activeSubscriptionNames = ConcurrentHashMap.newKeySet();
+ Set<String> failedSubscriptionNames = ConcurrentHashMap.newKeySet();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (Map.Entry<String, PersistentSubscription> subscriptionEntry : subscriptions.entrySet()) {
+ String subscriptionName = subscriptionEntry.getKey();
+ PersistentSubscription subscription = subscriptionEntry.getValue();
+ ManagedCursor cursor = subscription.getCursor();
+ if (!cursor.isDurable()) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ continue;
+ }
+ activeSubscriptionNames.add(subscriptionName);
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ if (markDeletePosition == null) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ continue;
+ }
+ if (isSubscriptionOldPositionInfoReusable(subscriptionName, markDeletePosition)) {
+ continue;
+ }
+ if (subscription.getNumberOfEntriesInBacklog(preciseTimeBasedBacklogQuotaCheck) <= 0) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ continue;
+ }
+
+ if (preciseTimeBasedBacklogQuotaCheck) {
+ futures.add(updateSubscriptionOldPositionInfoPrecise(
+ subscriptionName, subscription, cursor, markDeletePosition, dataVersion,
+ failedSubscriptionNames, preciseTimeBasedBacklogQuotaCheck));
+ } else {
+ try {
+ EstimateTimeBasedBacklogQuotaCheckResult checkResult =
+ estimatedTimeBasedBacklogQuotaCheck(markDeletePosition);
+ if (checkResult.getEstimatedOldestUnacknowledgedMessageTimestamp() != null) {
+ updateSubscriptionResultIfNewer(subscriptionName,
+ new OldestPositionInfo(
+ markDeletePosition,
+ cursor.getName(),
+ checkResult.getEstimatedOldestUnacknowledgedMessageTimestamp(),
+ dataVersion));
+ } else {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ failedSubscriptionNames.add(subscriptionName);
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ break;
+ } catch (ExecutionException e) {
+ failedSubscriptionNames.add(subscriptionName);
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ }
+ }
+ }
+
+ Map<String, OldestPositionInfo> positionInfos = subscriptionOldestPositionInfos;
+ if (positionInfos != null) {
+ positionInfos.keySet().removeIf(subscriptionName -> !activeSubscriptionNames.contains(subscriptionName));
+ }
+ if (futures.isEmpty()) {
+ logSubscriptionOldPositionInfoUpdateFailures(failedSubscriptionNames);
+ return CompletableFuture.completedFuture(null);
+ }
+ return FutureUtil.waitForAll(futures)
+ .thenRun(() -> logSubscriptionOldPositionInfoUpdateFailures(failedSubscriptionNames));
+ }
+
+ private void logSubscriptionOldPositionInfoUpdateFailures(Set<String> failedSubscriptionNames) {
+ if (!failedSubscriptionNames.isEmpty()) {
+ log.warn("Failed to update subscription old position info for {} subscriptions",
+ failedSubscriptionNames.size());
+ }
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfoPrecise(String subscriptionName,
+ PersistentSubscription subscription,
+ ManagedCursor cursor,
+ Position markDeletePosition,
+ long dataVersion,
+ Set<String> failedSubscriptionNames,
+ boolean preciseBacklogQuotaCheck) {
+ Position position = ledger.getNextValidPosition(markDeletePosition);
+ if (position.compareTo(ledger.getLastConfirmedEntry()) > 0) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ ledger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ if (isSubscriptionOldPositionInfoReadResultStillValid(
+ subscriptionName, subscription, cursor, markDeletePosition, dataVersion,
+ preciseBacklogQuotaCheck)) {
+ updateSubscriptionResultIfNewer(subscriptionName,
+ new OldestPositionInfo(
+ markDeletePosition,
+ cursor.getName(),
+ entry.getEntryTimestamp(),
+ dataVersion));
+ }
+ future.complete(null);
+ } catch (Exception e) {
+ failedSubscriptionNames.add(subscriptionName);
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ future.complete(null);
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ failedSubscriptionNames.add(subscriptionName);
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ future.complete(null);
+ }
+ }, null);
+ return future;
+ }
+
+ private boolean isSubscriptionOldPositionInfoReadResultStillValid(String subscriptionName,
+ PersistentSubscription subscription,
+ ManagedCursor cursor,
+ Position markDeletePosition,
+ long dataVersion,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ PersistentSubscription currentSubscription = subscriptions.get(subscriptionName);
+ if (currentSubscription != subscription || currentSubscription.getCursor() != cursor) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ return false;
+ }
+ Position currentMarkDeletePosition = cursor.getMarkDeletedPosition();
+ if (currentMarkDeletePosition == null) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ return false;
+ }
+ if (currentMarkDeletePosition.compareTo(markDeletePosition) != 0) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ return false;
+ }
+ if (subscription.getNumberOfEntriesInBacklog(preciseTimeBasedBacklogQuotaCheck) <= 0) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ return false;
+ }
+ return true;
+ }
+
+ public CompletableFuture<Void> updateSubscriptionOldPositionInfo() {
+ if (!brokerService.pulsar().getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus()) {
+ clearSubscriptionOldestPositionInfos();
+ return CompletableFuture.completedFuture(null);
+ }
+
+ if (!(ledger.getCursors() instanceof ManagedCursorContainer managedCursorContainer)) {
+ // TODO: support this method with a customized managed ledger implementation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ boolean preciseTimeBasedBacklogQuotaCheck =
+ brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck();
+ if (!hasBacklogs(preciseTimeBasedBacklogQuotaCheck)) {
+ clearSubscriptionOldestPositionInfos();
+ return CompletableFuture.completedFuture(null);
+ }
+ return updateSubscriptionOldPositionInfos(managedCursorContainer, preciseTimeBasedBacklogQuotaCheck);
+ }
+
public CompletableFuture<Void> updateOldPositionInfo() {
TopicName topicName = TopicName.get(getName());
@@ -3847,7 +4093,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return CompletableFuture.completedFuture(null);
}
- if (!hasBacklogs(brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck())) {
+ boolean preciseTimeBasedBacklogQuotaCheck =
+ brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck();
+ if (!hasBacklogs(preciseTimeBasedBacklogQuotaCheck)) {
if (log.isDebugEnabled()) {
log.debug("[{}] No backlog. Update old position info is null", topicName);
}
@@ -3855,7 +4103,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return CompletableFuture.completedFuture(null);
}
- // If we have no durable cursor since `ledger.getCursors()` only managed durable cursors
+ // If the oldest-cursor heap has no durable cursor, there is no topic-level oldest backlog position.
CursorInfo oldestMarkDeleteCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
if (oldestMarkDeleteCursorInfo == null || oldestMarkDeleteCursorInfo.getPosition() == null) {
if (log.isDebugEnabled()) {
@@ -3888,7 +4136,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
return CompletableFuture.completedFuture(null);
}
- if (brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck()) {
+ if (preciseTimeBasedBacklogQuotaCheck) {
CompletableFuture<Void> future = new CompletableFuture<>();
// Check if first unconsumed message(first message after mark delete position)
// for slowest cursor's has expired.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
index ca03e97c8396..aa6a71034147 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
@@ -29,6 +29,8 @@ public class AggregatedSubscriptionStats {
public long msgBacklogNoDelayed;
+ public long backlogAgeSeconds = -1;
+
public boolean blockedSubscriptionOnUnackedMsgs;
public double msgRateRedeliver;
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 dac48b9b1e5e..85027efe949b 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
@@ -86,6 +86,8 @@ public class NamespaceStatsAggregator {
Optional<CompactorMXBean> compactorMXBean = getCompactorMXBean(pulsar);
LongAdder topicsCount = new LongAdder();
Map<String, Long> localNamespaceTopicCount = new HashMap<>();
+ boolean exposeSubscriptionBacklogAge =
+ pulsar.getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus();
pulsar.getBrokerService().getMultiLayerTopicsMap().forEach((namespace, bundlesMap) -> {
namespaceStats.reset();
topicsCount.reset();
@@ -118,7 +120,7 @@ public class NamespaceStatsAggregator {
topicsCount.add(1);
TopicStats.printTopicStats(stream, topicStats, compactorMXBean, cluster, namespace, name,
- splitTopicAndPartitionIndexLabel, customLabelAndValues);
+ splitTopicAndPartitionIndexLabel, exposeSubscriptionBacklogAge, customLabelAndValues);
} else {
namespaceStats.updateStats(topicStats);
}
@@ -152,6 +154,7 @@ public class NamespaceStatsAggregator {
subsStats.bytesOutCounter = subscriptionStats.bytesOutCounter;
subsStats.msgOutCounter = subscriptionStats.msgOutCounter;
subsStats.msgBacklog = subscriptionStats.msgBacklog;
+ subsStats.backlogAgeSeconds = subscriptionStats.oldestBacklogMessageAgeSeconds;
subsStats.msgDelayed = subscriptionStats.msgDelayed;
subsStats.msgInReplay = subscriptionStats.msgInReplay;
subsStats.msgRateExpired = subscriptionStats.msgRateExpired;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
index d62993211c22..0c1d520d9bcc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
@@ -148,6 +148,7 @@ class TopicStats {
public static void printTopicStats(PrometheusMetricStreams stream, TopicStats stats,
Optional<CompactorMXBean> compactorMXBean, String cluster, String namespace,
String topic, boolean splitTopicAndPartitionIndexLabel,
+ boolean exposeSubscriptionBacklogAge,
String[] customLabelsAndValues) {
writeMetric(stream, "pulsar_subscriptions_count", stats.subscriptionsCount,
cluster, namespace, topic, splitTopicAndPartitionIndexLabel, customLabelsAndValues);
@@ -324,6 +325,11 @@ class TopicStats {
writeSubscriptionMetric(stream, "pulsar_subscription_back_log_no_delayed",
subsStats.msgBacklogNoDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel,
customLabelsAndValues);
+ if (exposeSubscriptionBacklogAge && subsStats.backlogAgeSeconds >= 0) {
+ writeSubscriptionMetric(stream, "pulsar_subscription_storage_backlog_age_seconds",
+ subsStats.backlogAgeSeconds, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel,
+ customLabelsAndValues);
+ }
writeSubscriptionMetric(stream, "pulsar_subscription_delayed",
subsStats.msgDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel,
customLabelsAndValues);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
index a693364ec8fa..3b6804a8d19f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
@@ -41,6 +41,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
@@ -78,6 +79,7 @@ import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.TopicType;
@@ -202,14 +204,28 @@ public class BacklogQuotaManagerTest {
}
@BeforeMethod(alwaysRun = true)
- void createNamespaces() throws PulsarAdminException {
+ void createNamespaces() throws Exception {
config.setPreciseTimeBasedBacklogQuotaCheck(false);
- admin.namespaces().createNamespace("prop/ns-quota");
- admin.namespaces().setNamespaceReplicationClusters("prop/ns-quota", Sets.newHashSet("usc"), false);
- admin.namespaces().createNamespace("prop/quotahold");
- admin.namespaces().setNamespaceReplicationClusters("prop/quotahold", Sets.newHashSet("usc"), false);
- admin.namespaces().createNamespace("prop/quotaholdasync");
- admin.namespaces().setNamespaceReplicationClusters("prop/quotaholdasync", Sets.newHashSet("usc"), false);
+ config.setExposeSubscriptionBacklogAgeInPrometheus(false);
+ createNamespaceForTest("prop/ns-quota");
+ createNamespaceForTest("prop/quotahold");
+ createNamespaceForTest("prop/quotaholdasync");
+ }
+
+ /**
+ * If a previous test's @AfterMethod timed out before the namespace was fully removed, the
+ * leftover would otherwise cascade as HTTP 409 here and fail every subsequent test.
+ * Force-delete and retry so each test starts with clean namespace state.
+ */
+ private void createNamespaceForTest(String ns) throws Exception {
+ try {
+ admin.namespaces().createNamespace(ns);
+ } catch (PulsarAdminException.ConflictException e) {
+ log.warn("Namespace {} already exists from previous test, force-deleting and recreating", ns);
+ deleteNamespaceWithRetry(ns, true);
+ admin.namespaces().createNamespace(ns);
+ }
+ admin.namespaces().setNamespaceReplicationClusters(ns, Sets.newHashSet("usc"), false);
}
@AfterMethod(alwaysRun = true)
@@ -314,6 +330,12 @@ public class BacklogQuotaManagerTest {
return stats;
}
+ private void refreshSubscriptionBacklogAge(String topic) throws Exception {
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+ }
+
+ @SuppressWarnings("deprecation")
@Test
public void testTriggerBacklogQuotaSizeWithReader() throws Exception {
assertEquals(admin.namespaces().getBacklogQuotaMap("prop/ns-quota"),
@@ -385,8 +407,9 @@ public class BacklogQuotaManagerTest {
}
@Test
- public void backlogsStatsPrecise() throws PulsarAdminException, PulsarClientException, InterruptedException {
+ public void backlogsStatsPrecise() throws Exception {
config.setPreciseTimeBasedBacklogQuotaCheck(true);
+ config.setExposeSubscriptionBacklogAgeInPrometheus(true);
final String namespace = "prop/ns-quota";
assertEquals(admin.namespaces().getBacklogQuotaMap(namespace), new HashMap<>());
final int sizeLimitBytes = 15 * 1024 * 1024;
@@ -443,6 +466,7 @@ public class BacklogQuotaManagerTest {
c1MarkDeletePositionBefore = waitForMarkDeletePositionToChange(topic1, subName1,
c1MarkDeletePositionBefore);
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
Metrics metrics = prometheusMetricsClient.getMetrics();
TopicStats topicStats = getTopicStats(topic1);
@@ -454,6 +478,11 @@ public class BacklogQuotaManagerTest {
assertThat(topicStats.getOldestBacklogMessageAgeSeconds())
.isCloseTo(expectedMessageAgeSeconds, within(1L));
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isCloseTo(expectedMessageAgeSeconds, within(1L));
+ assertThat(topicStats.getSubscriptions().get(subName1).getOldestBacklogMessageAgeSeconds())
+ .isGreaterThanOrEqualTo(0L)
+ .isLessThan(expectedMessageAgeSeconds);
Metric backlogAgeMetric =
metrics.findSingleMetricByNameAndLabels("pulsar_storage_backlog_age_seconds",
@@ -463,6 +492,15 @@ public class BacklogQuotaManagerTest {
entry("namespace", namespace),
entry("topic", topic1));
assertThat((long) backlogAgeMetric.value).isCloseTo(expectedMessageAgeSeconds, within(2L));
+ Metric subBacklogAgeMetric =
+ metrics.findSingleMetricByNameAndLabels("pulsar_subscription_storage_backlog_age_seconds",
+ Pair.of("topic", topic1), Pair.of("subscription", subName2));
+ assertThat(subBacklogAgeMetric.tags).containsExactly(
+ entry("cluster", CLUSTER_NAME),
+ entry("namespace", namespace),
+ entry("subscription", subName2),
+ entry("topic", topic1));
+ assertThat((long) subBacklogAgeMetric.value).isCloseTo(expectedMessageAgeSeconds, within(2L));
// Move subscription 2 away from being the oldest mark delete
// S2/S1
@@ -482,6 +520,7 @@ public class BacklogQuotaManagerTest {
c1MarkDeletePositionBefore = waitForMarkDeletePositionToChange(topic1, subName1,
c1MarkDeletePositionBefore);
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
metrics = prometheusMetricsClient.getMetrics();
long actualAge = (long) metrics.findByNameAndLabels(
@@ -508,6 +547,7 @@ public class BacklogQuotaManagerTest {
waitForMarkDeletePositionToChange(topic1, subName1,
c1MarkDeletePositionBefore);
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
// Cache shouldn't be used, since position has changed
long readEntries = getReadEntries(topic1);
@@ -518,8 +558,11 @@ public class BacklogQuotaManagerTest {
MILLISECONDS.toSeconds(System.currentTimeMillis() - secondOldestMessage.getPublishTime());
assertThat(topicStats.getOldestBacklogMessageAgeSeconds()).isCloseTo(expectedMessageAgeSeconds, within(2L));
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isCloseTo(expectedMessageAgeSeconds, within(2L));
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
// Cache should be used, since position hasn't changed
assertThat(getReadEntries(topic1)).isEqualTo(readEntries);
@@ -537,11 +580,14 @@ public class BacklogQuotaManagerTest {
log.info("Subscription 1 and 2 moved to end. Now should not backlog");
waitForMarkDeletePositionToChange(topic1, subName1, c1MarkDeletePositionBefore);
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
topicStats = getTopicStats(topic1);
assertThat(topicStats.getBacklogSize()).isEqualTo(0);
assertThat(topicStats.getSubscriptions().get(subName1).getMsgBacklog()).isEqualTo(0);
assertThat(topicStats.getSubscriptions().get(subName2).getMsgBacklog()).isEqualTo(0);
+ assertThat(topicStats.getSubscriptions().get(subName1).getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
assertThat(topicStats.getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isNull();
@@ -561,6 +607,117 @@ public class BacklogQuotaManagerTest {
}
}
+ @Test
+ public void subscriptionBacklogAgeStatsDisabledByDefault() throws Exception {
+ config.setPreciseTimeBasedBacklogQuotaCheck(true);
+
+ try (PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
+ .statsInterval(0, SECONDS).build()) {
+ final String topic1 = "persistent://prop/ns-quota/topic-disabled" + UUID.randomUUID();
+ final String subName = "c1";
+
+ client.newConsumer().topic(topic1).subscriptionName(subName)
+ .acknowledgmentGroupTime(0, SECONDS)
+ .subscribe();
+ Producer<byte[]> producer = createProducer(client, topic1);
+ producer.send(new byte[1024]);
+
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic1).get();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+
+ TopicStats topicStats = getTopicStats(topic1);
+ assertThat(topicStats.getSubscriptions().get(subName).getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
+
+ Metrics metrics = prometheusMetricsClient.getMetrics();
+ assertThat(metrics.findByNameAndLabels("pulsar_subscription_storage_backlog_age_seconds",
+ Pair.of("topic", topic1), Pair.of("subscription", subName))).isEmpty();
+ }
+ }
+
+ @Test
+ public void subscriptionBacklogAgeSkipsNonDurableReader() throws Exception {
+ config.setPreciseTimeBasedBacklogQuotaCheck(true);
+ config.setExposeSubscriptionBacklogAgeInPrometheus(true);
+
+ try (PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
+ .statsInterval(0, SECONDS).build()) {
+ final String topic1 = "persistent://prop/ns-quota/topic-reader" + UUID.randomUUID();
+ final String subName = "c1";
+
+ client.newConsumer().topic(topic1).subscriptionName(subName)
+ .acknowledgmentGroupTime(0, SECONDS)
+ .subscribe();
+ client.newReader().topic(topic1).startMessageId(MessageId.earliest).create();
+ Producer<byte[]> producer = createProducer(client, topic1);
+ producer.send(new byte[1024]);
+
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic1).get();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+
+ TopicStats topicStats = getTopicStats(topic1);
+ assertThat(topicStats.getSubscriptions().get(subName).getOldestBacklogMessageAgeSeconds())
+ .isGreaterThanOrEqualTo(0L);
+
+ Map.Entry<String, ? extends SubscriptionStats> readerStats = topicStats.getSubscriptions().entrySet()
+ .stream()
+ .filter(entry -> !entry.getValue().isDurable())
+ .findFirst()
+ .orElseThrow();
+ assertThat(readerStats.getValue().getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
+
+ Metrics metrics = prometheusMetricsClient.getMetrics();
+ assertThat(metrics.findByNameAndLabels("pulsar_subscription_storage_backlog_age_seconds",
+ Pair.of("topic", topic1), Pair.of("subscription", readerStats.getKey()))).isEmpty();
+ }
+ }
+
+ @Test
+ public void subscriptionBacklogAgeCacheRemovedAfterSubscriptionDeleteAndRecreate() throws Exception {
+ config.setPreciseTimeBasedBacklogQuotaCheck(true);
+ config.setExposeSubscriptionBacklogAgeInPrometheus(true);
+
+ try (PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
+ .statsInterval(0, SECONDS).build()) {
+ final String topic1 = "persistent://prop/ns-quota/topic-recreated" + UUID.randomUUID();
+ final String subName1 = "c1";
+ final String subName2 = "c2";
+
+ Consumer<byte[]> consumer1 = client.newConsumer().topic(topic1).subscriptionName(subName1)
+ .acknowledgmentGroupTime(0, SECONDS)
+ .subscribe();
+ client.newConsumer().topic(topic1).subscriptionName(subName2)
+ .acknowledgmentGroupTime(0, SECONDS)
+ .subscribe();
+ Producer<byte[]> producer = createProducer(client, topic1);
+ producer.send(new byte[1024]);
+
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic1).get();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+
+ TopicStats topicStats = getTopicStats(topic1);
+ assertThat(topicStats.getSubscriptions().get(subName1).getOldestBacklogMessageAgeSeconds())
+ .isGreaterThanOrEqualTo(0L);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isGreaterThanOrEqualTo(0L);
+
+ consumer1.unsubscribe();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+
+ client.newConsumer().topic(topic1).subscriptionName(subName1)
+ .acknowledgmentGroupTime(0, SECONDS)
+ .subscribe();
+ topicRef.updateSubscriptionOldPositionInfo().get(5, SECONDS);
+
+ topicStats = getTopicStats(topic1);
+ assertThat(topicStats.getSubscriptions().get(subName1).getMsgBacklog()).isEqualTo(0);
+ assertThat(topicStats.getSubscriptions().get(subName1).getOldestBacklogMessageAgeSeconds()).isEqualTo(-1);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isGreaterThanOrEqualTo(0L);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+
@Test
public void backlogsStatsPreciseWithNoBacklog() throws PulsarAdminException,
PulsarClientException, InterruptedException {
@@ -782,8 +939,9 @@ public class BacklogQuotaManagerTest {
}
@Test
- public void backlogsStatsNotPrecise() throws PulsarAdminException, PulsarClientException, InterruptedException {
+ public void backlogsStatsNotPrecise() throws Exception {
config.setPreciseTimeBasedBacklogQuotaCheck(false);
+ config.setExposeSubscriptionBacklogAgeInPrometheus(true);
config.setManagedLedgerMaxEntriesPerLedger(6);
final String namespace = "prop/ns-quota";
assertEquals(admin.namespaces().getBacklogQuotaMap(namespace), new HashMap<>());
@@ -852,12 +1010,15 @@ public class BacklogQuotaManagerTest {
long unloadTime = System.currentTimeMillis();
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
topicStats = getTopicStats(topic1);
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
// age is measured against the ledger closing time
long expectedAge = MILLISECONDS.toSeconds(System.currentTimeMillis() - unloadTime);
assertThat(topicStats.getOldestBacklogMessageAgeSeconds()).isCloseTo(expectedAge, within(1L));
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isCloseTo(expectedAge, within(1L));
String c2MarkDeletePositionBefore =
admin.topics().getInternalStats(topic1).cursors.get(subName2).markDeletePosition;
@@ -872,11 +1033,14 @@ public class BacklogQuotaManagerTest {
waitForMarkDeletePositionToChange(topic1, subName1, c1MarkDeletePositionBefore);
waitForMarkDeletePositionToChange(topic1, subName2, c2MarkDeletePositionBefore);
waitForQuotaCheckToRunTwice();
+ refreshSubscriptionBacklogAge(topic1);
topicStats = getTopicStats(topic1);
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
expectedAge = MILLISECONDS.toSeconds(System.currentTimeMillis() - unloadTime);
assertThat(topicStats.getOldestBacklogMessageAgeSeconds()).isCloseTo(expectedAge, within(1L));
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
+ .isCloseTo(expectedAge, within(1L));
// Unsubscribe consume1 and consumer2
consumer1.unsubscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
index e51d6a886e01..7f776d862d68 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
@@ -100,7 +100,8 @@ public class NamespaceStatsAggregatorTest {
when(topic.getSubscriptions()).thenReturn(subscriptionsMaps);
when(topic.getReplicators()).thenReturn(new ConcurrentHashMap<>());
when(topic.getManagedLedger()).thenReturn(ml);
- when(topic.getBacklogQuota(Mockito.any())).thenReturn(Mockito.mock(BacklogQuota.class));
+ BacklogQuota backlogQuota = Mockito.mock(BacklogQuota.class);
+ when(topic.getBacklogQuota(Mockito.any())).thenReturn(backlogQuota);
PersistentTopicMetrics persistentTopicMetrics = new PersistentTopicMetrics();
when(topic.getPersistentTopicMetrics()).thenReturn(persistentTopicMetrics);
topicsMap.put("my-topic", topic);
@@ -108,7 +109,10 @@ public class NamespaceStatsAggregatorTest {
PrometheusMetricStreams metricStreams = Mockito.spy(new PrometheusMetricStreams());
// Populate subscriptions stats
+ ServiceConfiguration config = pulsar.getConfiguration();
+ doReturn(true).when(config).isExposeSubscriptionBacklogAgeInPrometheus();
subStats.blockedSubscriptionOnUnackedMsgs = true;
+ subStats.oldestBacklogMessageAgeSeconds = 123;
consumerStats.blockedConsumerOnUnackedMsgs = false; // should not affect blockedSubscriptionOnUnackedMsgs
consumerStats.unackedMessages = 1;
consumerStats.msgRateRedeliver = 0.7;
@@ -124,6 +128,7 @@ public class NamespaceStatsAggregatorTest {
verifySubscriptionMetric(metricStreams, "pulsar_subscription_msg_rate_redeliver", 0.7);
verifySubscriptionMetric(metricStreams, "pulsar_subscription_unacked_messages", 1L);
+ verifySubscriptionMetric(metricStreams, "pulsar_subscription_storage_backlog_age_seconds", 123L);
}
private void verifySubscriptionMetric(PrometheusMetricStreams metricStreams, String metricName, Number value) {
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
index 05cfd55b0456..aa26c8bc40d9 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
@@ -57,6 +57,18 @@ public interface SubscriptionStats {
/** Get the publish time of the earliest message in the backlog. */
long getEarliestMsgPublishTimeInBacklog();
+ /**
+ * Age of oldest unacknowledged message for this subscription, in seconds.
+ * <p>
+ * This is a best-effort cached value from the broker's periodic subscription backlog-age refresh. The value is
+ * {@code -1} when it is unknown, not applicable, the subscription has no backlog, or the broker has disabled
+ * subscription backlog-age computation.
+ * </p>
+ */
+ default long getOldestBacklogMessageAgeSeconds() {
+ return -1;
+ }
+
/** Number of entries in the subscription backlog that do not contain the delay messages. */
long getMsgBacklogNoDelayed();
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
index efe53bba7d6e..3939939037ca 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
@@ -65,6 +65,16 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
/** Get the publish time of the earliest message in the backlog. */
public long earliestMsgPublishTimeInBacklog;
+ /**
+ * Age of oldest unacknowledged message for this subscription, in seconds.
+ * <p>
+ * This is a best-effort cached value from the broker's periodic subscription backlog-age refresh. The value is
+ * {@code -1} when it is unknown, not applicable, the subscription has no backlog, or the broker has disabled
+ * subscription backlog-age computation.
+ * </p>
+ */
+ public long oldestBacklogMessageAgeSeconds = -1;
+
/** Number of entries in the subscription backlog that do not contain the delay messages. */
public long msgBacklogNoDelayed;
@@ -224,6 +234,7 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
nonContiguousDeletedMessagesRanges = 0;
nonContiguousDeletedMessagesRangesSerializedSize = 0;
earliestMsgPublishTimeInBacklog = 0L;
+ oldestBacklogMessageAgeSeconds = -1;
delayedMessageIndexSizeInBytes = 0;
subscriptionProperties.clear();
filterProcessedMsgCount = 0;
@@ -290,6 +301,8 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
stats.earliestMsgPublishTimeInBacklog
);
}
+ this.oldestBacklogMessageAgeSeconds = Math.max(
+ this.oldestBacklogMessageAgeSeconds, stats.oldestBacklogMessageAgeSeconds);
this.delayedMessageIndexSizeInBytes += stats.delayedMessageIndexSizeInBytes;
this.subscriptionProperties.putAll(stats.subscriptionProperties);
this.filterProcessedMsgCount += stats.filterProcessedMsgCount;
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImplTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImplTest.java
index 8a4b5da9edd2..d6c3222d2b01 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImplTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImplTest.java
@@ -27,8 +27,10 @@ public class SubscriptionStatsImplTest {
public void testReset() {
SubscriptionStatsImpl stats = new SubscriptionStatsImpl();
stats.earliestMsgPublishTimeInBacklog = 1L;
+ stats.oldestBacklogMessageAgeSeconds = 10L;
stats.reset();
assertEquals(stats.earliestMsgPublishTimeInBacklog, 0L);
+ assertEquals(stats.oldestBacklogMessageAgeSeconds, -1L);
}
@@ -79,4 +81,19 @@ public class SubscriptionStatsImplTest {
SubscriptionStatsImpl aggregate = stats1.add(stats2);
assertEquals(aggregate.earliestMsgPublishTimeInBacklog, 0L);
}
-}
\ No newline at end of file
+
+ @Test
+ public void testAdd_OldestBacklogMessageAgeSeconds() {
+ SubscriptionStatsImpl stats1 = new SubscriptionStatsImpl();
+ stats1.oldestBacklogMessageAgeSeconds = -1L;
+
+ SubscriptionStatsImpl stats2 = new SubscriptionStatsImpl();
+ stats2.oldestBacklogMessageAgeSeconds = 20L;
+
+ SubscriptionStatsImpl stats3 = new SubscriptionStatsImpl();
+ stats3.oldestBacklogMessageAgeSeconds = 10L;
+
+ SubscriptionStatsImpl aggregate = stats1.add(stats2).add(stats3);
+ assertEquals(aggregate.oldestBacklogMessageAgeSeconds, 20L);
+ }
+}