[improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric

conf/broker.conf

@@ -1973,6 +1973,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

conf/standalone.conf

@@ -1249,6 +1249,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).

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -3837,6 +3837,15 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
     )
     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"

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -259,6 +259,7 @@ public class BrokerService implements Closeable {
 
     @Getter
     private final SingleThreadNonConcurrentFixedRateScheduler backlogQuotaChecker;
+    private final SingleThreadNonConcurrentFixedRateScheduler subscriptionBacklogAgeChecker;
 
     protected final AtomicReference<Semaphore> lookupRequestSemaphore;
     @Getter
@@ -402,6 +403,8 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
                 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();
@@ -692,6 +695,7 @@ public void start() throws Exception {
         this.startCompactionMonitor();
         this.startConsumedLedgersMonitor();
         this.startBacklogQuotaChecker();
+        this.startSubscriptionBacklogAgeChecker();
         this.updateBrokerPublisherThrottlingMaxRate();
         this.updateBrokerDispatchThrottlingMaxRate();
         this.startCheckReplicationPolicies();
@@ -890,6 +894,19 @@ protected void startBacklogQuotaChecker() {
 
     }
 
+    protected void startSubscriptionBacklogAgeChecker() {
+        if (pulsar().getConfiguration().isExposeSubscriptionBacklogAgeInPrometheus()) {
+            final int interval = pulsar().getConfiguration().getBacklogQuotaCheckIntervalInSeconds();
+            log.info()
+                    .attr("intervalSeconds", interval)
+                    .log("Scheduling a thread to refresh subscription backlog age in background");
+            subscriptionBacklogAgeChecker.scheduleAtFixedRateNonConcurrently(
+                    () -> refreshSubscriptionBacklogAge().join(), interval, interval, TimeUnit.SECONDS);
+        } else {
+            log.info("Subscription backlog age monitoring is disabled");
+        }
+    }
+
     public void close() throws IOException {
         try {
             closeAsync().get();
@@ -1043,6 +1060,7 @@ public CompletableFuture<Void> closeAsync() {
                                                 compactionMonitor,
                                                 consumedLedgersMonitor,
                                                 backlogQuotaChecker,
+                                                subscriptionBacklogAgeChecker,
                                                 topicOrderedExecutor,
                                                 deduplicationSnapshotMonitor,
                                                 segmentLoadReporterMonitor)
@@ -2586,6 +2604,21 @@ public void monitorBacklogQuota() {
         });
     }
 
+    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()
+                    .exception(throwable)
+                    .log("Error when refreshSubscriptionBacklogAge()");
+            return null;
+        });
+    }
+
     public CompletableFuture<Boolean> isTopicNsOwnedByBrokerAsync(TopicName topicName) {
         return pulsar.getNamespaceService().isServiceUnitOwnedAsync(topicName)
                 .handle((hasOwnership, t) -> {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -1514,6 +1514,8 @@ public CompletableFuture<SubscriptionStatsImpl> getStatsAsync(GetStatsOptions ge
             }
         }
         subStats.msgBacklog = getNumberOfEntriesInBacklog(getStatsOptions.isGetPreciseBacklog());
+        subStats.oldestBacklogMessageAgeSeconds =
+                topic.getBestEffortOldestUnacknowledgedMessageAgeSeconds(subName);
         if (getStatsOptions.isSubscriptionBacklogSize()) {
             subStats.backlogSize = topic.getManagedLedger()
                     .getEstimatedBacklogSize(cursor.getMarkDeletedPosition());