[improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric

[Technoboy-]

Motivation

Pulsar currently exposes pulsar_storage_backlog_age_seconds to report the age of the oldest unacknowledged message for a topic. This metric is useful for detecting consumer lag based on actual message age instead of only backlog size.

However, the metric is only available at the topic level. When a topic has multiple subscriptions, operators cannot tell which subscription is contributing to the backlog age. This is a common production case because different subscriptions on the same topic can have different latency expectations. Some subscriptions are expected to consume messages in real time, while others may intentionally retain backlog for a longer period.

Without a subscription-level backlog age metric, users have to either alert on the topic-level metric and tolerate false positives, or build complex Prometheus rules to exclude known slow subscriptions. This makes monitoring harder and less accurate.

This change adds pulsar_subscription_storage_backlog_age_seconds, allowing users to monitor backlog age per subscription and configure alerts based on each subscription's expected latency.

Modifications

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[Copilot]

Pull request overview

This PR adds a subscription-level Prometheus metric, pulsar_subscription_storage_backlog_age_seconds, to help operators identify which subscription is driving backlog age (consumer lag) on multi-subscription topics.

Changes:

• Extend subscription stats with oldestBacklogMessageAgeSeconds and aggregation/reset behavior.
• Compute per-subscription “best-effort oldest unacked message age” in the broker and propagate it into Prometheus subscription metrics.
• Add unit/test coverage for stats reset/aggregation and Prometheus metric emission.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImplTest.java	Adds tests for reset and aggregation behavior of the new subscription backlog-age stat.
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java	Introduces oldestBacklogMessageAgeSeconds, resets it, and aggregates it via max.
pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java	Adds the new getOldestBacklogMessageAgeSeconds() accessor to the subscription stats API.
pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java	Verifies the new pulsar_subscription_storage_backlog_age_seconds metric is emitted.
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java	Emits the new subscription-level backlog-age metric in Prometheus output.
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java	Maps subscription stats’ oldestBacklogMessageAgeSeconds into aggregated subscription metric state.
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java	Adds storage for per-subscription backlog age used by Prometheus output.
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java	Adds per-subscription cached oldest-position tracking and an overload to read per-subscription backlog age.
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java	Plumbs per-subscription backlog age into SubscriptionStatsImpl.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[lhotari]

Thanks for working on this — per-subscription backlog age is genuinely useful, and the cache design mirrors the topic-level one closely.

My main request is a dedicated configuration key, defaulting to false, that gates the computation, not just the metric emission (details inline at PersistentTopic.java:4127). Beyond that I found several correctness issues that I think need fixing regardless of the flag — the most significant being that non-durable (Reader) subscriptions report a wrong value in two easily-reachable topologies.

Why the flag

updateSubscriptionOldPositionInfos() is called unconditionally, so every persistent topic with any backlog pays a full walk of every cursor every backlogQuotaCheckIntervalInSeconds (default 60) on the single pulsar-backlog-quota-checker thread — including when no backlog quota is configured at all.

To be precise about what is new: the pass already began with hasBacklogs(), which is a short-circuiting anyMatch over subscriptions, and then did one estimate — or one asyncReadEntry for the single oldest cursor — for the topic. What this PR adds is a full, non-short-circuiting walk of the whole ManagedCursorContainer, with a hasBacklog() call for every cursor rather than up to the first backlogged one, plus an estimate or read per backlogged cursor whose mark-delete moved.

Meanwhile the metric is already gated. Emission goes through TopicStats.printTopicStats, which NamespaceStatsAggregator only reaches when exposeTopicLevelMetricsInPrometheus is true. Operators who turn topic-level metrics off — typically the large clusters — would pay the whole cost and never scrape the metric. (The cache also backs an ungated admin-API field, SubscriptionStats.oldestBacklogMessageAgeSeconds, so a default-false flag would blank that to -1 as well. I think that is the right trade, but it should be a conscious one.) The only existing knob is backlogQuotaCheckEnabled=false, and it only stops the periodic checker — producer-creation checks still trigger the same walk.

PIP-323 justified the topic-level metric as "cheap to retrieve (no additional I/O involved)" precisely because it piggybacks on the oldest-cursor calculation the checker already had to do, and its Alternatives section rejected a costlier variant because it "might result in more frequent I/O calls, especially with many topics". This change breaks that property.

There is good precedent for default-off on expensive per-entity stats: exposePreciseBacklogInPrometheus, exposeSubscriptionBacklogSizeInPrometheus, exposeConsumerLevelMetricsInPrometheus and exposeManagedCursorMetricsInPrometheus are all false by default.

In fairness to the design, the reusable-position check on line 4023 blunts the steady state: an idle, abandoned backlogged subscription is a cache hit forever after the first pass, which is the main case this metric exists for, so it costs no repeat read. And reads go through InflightReadsLimiter, so they are not unbounded — though they now contend with consumer reads for that shared permit budget.

Minor

• Map<String, Boolean> activeSubscriptionNames at PersistentTopic.java:4010 could be a Set<String>.
• The catch (Exception e) at line 4044 swallows InterruptedException without restoring the interrupt flag.
• The PR description still has the template Modifications and Verifying this change sections unfilled, and the public API / metrics boxes unchecked — this does add a public admin-api method and a new metric, so a docs PR against apache/pulsar-site will be needed.

[Technoboy-]

Thanks for working on this — per-subscription backlog age is genuinely useful, and the cache design mirrors the topic-level one closely.

My main request is a dedicated configuration key, defaulting to false, that gates the computation, not just the metric emission (details inline at PersistentTopic.java:4127). Beyond that I found several correctness issues that I think need fixing regardless of the flag — the most significant being that non-durable (Reader) subscriptions report a wrong value in two easily-reachable topologies.

Why the flag

updateSubscriptionOldPositionInfos() is called unconditionally, so every persistent topic with any backlog pays a full walk of every cursor every backlogQuotaCheckIntervalInSeconds (default 60) on the single pulsar-backlog-quota-checker thread — including when no backlog quota is configured at all.

To be precise about what is new: the pass already began with hasBacklogs(), which is a short-circuiting anyMatch over subscriptions, and then did one estimate — or one asyncReadEntry for the single oldest cursor — for the topic. What this PR adds is a full, non-short-circuiting walk of the whole ManagedCursorContainer, with a hasBacklog() call for every cursor rather than up to the first backlogged one, plus an estimate or read per backlogged cursor whose mark-delete moved.

Meanwhile the metric is already gated. Emission goes through TopicStats.printTopicStats, which NamespaceStatsAggregator only reaches when exposeTopicLevelMetricsInPrometheus is true. Operators who turn topic-level metrics off — typically the large clusters — would pay the whole cost and never scrape the metric. (The cache also backs an ungated admin-API field, SubscriptionStats.oldestBacklogMessageAgeSeconds, so a default-false flag would blank that to -1 as well. I think that is the right trade, but it should be a conscious one.) The only existing knob is backlogQuotaCheckEnabled=false, and it only stops the periodic checker — producer-creation checks still trigger the same walk.

PIP-323 justified the topic-level metric as "cheap to retrieve (no additional I/O involved)" precisely because it piggybacks on the oldest-cursor calculation the checker already had to do, and its Alternatives section rejected a costlier variant because it "might result in more frequent I/O calls, especially with many topics". This change breaks that property.

There is good precedent for default-off on expensive per-entity stats: exposePreciseBacklogInPrometheus, exposeSubscriptionBacklogSizeInPrometheus, exposeConsumerLevelMetricsInPrometheus and exposeManagedCursorMetricsInPrometheus are all false by default.

In fairness to the design, the reusable-position check on line 4023 blunts the steady state: an idle, abandoned backlogged subscription is a cache hit forever after the first pass, which is the main case this metric exists for, so it costs no repeat read. And reads go through InflightReadsLimiter, so they are not unbounded — though they now contend with consumer reads for that shared permit budget.

Minor

• Map<String, Boolean> activeSubscriptionNames at PersistentTopic.java:4010 could be a Set<String>.
• The catch (Exception e) at line 4044 swallows InterruptedException without restoring the interrupt flag.
• The PR description still has the template Modifications and Verifying this change sections unfilled, and the public API / metrics boxes unchecked — this does add a public admin-api method and a new metric, so a docs PR against apache/pulsar-site will be needed.

please help review again