You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Technoboy- (via GitHub)" <gi...@apache.org> on 2026/08/11 14:01:40 UTC
[PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Technoboy- opened a new pull request, #26313:
URL: https://github.com/apache/pulsar/pull/26313
### 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
<!-- Describe the modifications you've done. -->
### 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:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3794119753
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java:
##########
@@ -324,6 +324,9 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st
writeSubscriptionMetric(stream, "pulsar_subscription_back_log_no_delayed",
subsStats.msgBacklogNoDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel,
customLabelsAndValues);
+ writeSubscriptionMetric(stream, "pulsar_subscription_storage_backlog_age_seconds",
Review Comment:
improved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#issuecomment-5421119310
It may be worth adding top-level exception handling around refreshSubscriptionBacklogAge().join(). Since scheduleAtFixedRateNonConcurrently does not wrap the task with SafeRunnable, an unexpected unchecked exception could prevent subsequent executions from being requeued. Adding a guard here would make the periodic refresh more resilient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#issuecomment-5377531298
> Thanks for the update. For this metric, caching by subscription and mark-delete position makes sense, especially for inactive subscriptions with long-lived backlog.
>
> I have three remaining suggestions:
>
> 1. Move subscription backlog-age refresh out of `updateOldPositionInfo()`.
> `updateOldPositionInfo()` is part of the backlog quota path, and can also be reached during producer creation through message-age backlog quota checks. With this change enabled, those paths may wait for per-subscription reads, even though subscription backlog age is only a best-effort metric.
> It would be cleaner to keep `updateOldPositionInfo()` focused on the existing topic-level oldest-position cache used by quota logic, and refresh the subscription-level cache from a separate background task. Prometheus/admin stats can then read the cache only.
> 2. Bound the background refresh with a simple broker-level concurrency limiter.
> This metric may need to read one entry per subscription when the cached mark-delete position changes. On a broker with many topics or many subscriptions, refreshing everything at once could create a burst of BookKeeper/offload reads and compete with normal consumer reads.
> A simple broker-level limiter should be enough. If no permit is available, skip that subscription and retry in a later refresh. Since this is a best-effort metric, skipping is better than queueing unbounded work or adding read pressure.
> 3. Re-check state before writing async read results.
> Async reads can complete after the subscription state has already changed. For example, a subscription may catch up, be deleted, or be recreated while the read is still in flight. In that case, the old result should not be written back into the cache.
> Before updating the cache, the callback should verify that the subscription still exists, the cursor is the same one, the mark-delete position is unchanged, and the subscription still has backlog. Otherwise, drop the result.
thanks for the good suggestions, addressed 1 and 3 for this patch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#issuecomment-5322396547
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3793928046
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3974,18 +4114,25 @@ public CompletableFuture<Void> updateOldPositionInfo() {
return CompletableFuture.completedFuture(null);
}
- if (!hasBacklogs(brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck())) {
+ boolean preciseTimeBasedBacklogQuotaCheck =
+ brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck();
+ if (!hasBacklogs(preciseTimeBasedBacklogQuotaCheck)) {
log.debug("No backlog. Update old position info is null");
TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.set(this, null);
+ subscriptionOldestPositionInfos.clear();
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> subscriptionOldPositionInfoUpdateFuture =
+ updateSubscriptionOldPositionInfos(managedCursorContainer, preciseTimeBasedBacklogQuotaCheck);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3794141112
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java:
##########
@@ -472,6 +472,8 @@ public void backlogsStatsPrecise() throws PulsarAdminException, PulsarClientExce
assertThat(topicStats.getOldestBacklogMessageAgeSeconds())
.isCloseTo(expectedMessageAgeSeconds, within(1L));
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
Review Comment:
improved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#issuecomment-5377085064
Thanks for the update. For this metric, caching by subscription and mark-delete position makes sense, especially for inactive subscriptions with long-lived backlog.
I have three remaining suggestions:
1. Move subscription backlog-age refresh out of `updateOldPositionInfo()`.
`updateOldPositionInfo()` is part of the backlog quota path, and can also be reached during producer creation through message-age backlog quota checks. With this change enabled, those paths may wait for per-subscription reads, even though subscription backlog age is only a best-effort metric.
It would be cleaner to keep `updateOldPositionInfo()` focused on the existing topic-level oldest-position cache used by quota logic, and refresh the subscription-level cache from a separate background task. Prometheus/admin stats can then read the cache only.
2. Bound the background refresh with a simple broker-level concurrency limiter.
This metric may need to read one entry per subscription when the cached mark-delete position changes. On a broker with many topics or many subscriptions, refreshing everything at once could create a burst of BookKeeper/offload reads and compete with normal consumer reads.
A simple broker-level limiter should be enough. If no permit is available, skip that subscription and retry in a later refresh. Since this is a best-effort metric, skipping is better than queueing unbounded work or adding read pressure.
3. Re-check state before writing async read results.
Async reads can complete after the subscription state has already changed. For example, a subscription may catch up, be deleted, or be recreated while the read is still in flight. In that case, the old result should not be written back into the cache.
Before updating the cache, the callback should verify that the subscription still exists, the cursor is the same one, the mark-delete position is unchanged, and the subscription still has backlog. Otherwise, drop the result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3794113002
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java:
##########
@@ -57,6 +57,16 @@ public interface SubscriptionStats {
/** Get the publish time of the earliest message in the backlog. */
long getEarliestMsgPublishTimeInBacklog();
+ /**
+ * Age of oldest unacknowledged message, as recorded in last backlog quota check interval.
+ * <p>
+ * The age of the oldest unacknowledged (i.e. backlog) message for this subscription, measured by the time elapsed
+ * from its published time, in seconds. This value is recorded every backlog quota check interval, hence it
+ * represents the value seen in the last check.
+ * </p>
+ */
+ long getOldestBacklogMessageAgeSeconds();
Review Comment:
addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3785528649
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3974,18 +4114,25 @@ public CompletableFuture<Void> updateOldPositionInfo() {
return CompletableFuture.completedFuture(null);
}
- if (!hasBacklogs(brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck())) {
+ boolean preciseTimeBasedBacklogQuotaCheck =
+ brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck();
+ if (!hasBacklogs(preciseTimeBasedBacklogQuotaCheck)) {
log.debug("No backlog. Update old position info is null");
TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.set(this, null);
+ subscriptionOldestPositionInfos.clear();
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> subscriptionOldPositionInfoUpdateFuture =
+ updateSubscriptionOldPositionInfos(managedCursorContainer, preciseTimeBasedBacklogQuotaCheck);
Review Comment:
**This is my main request: please gate this behind a new configuration key that defaults to `false`.**
The call is unconditional, so every persistent topic with any backlog pays the full per-cursor walk every `backlogQuotaCheckIntervalInSeconds` (default 60) on the single `pulsar-backlog-quota-checker` thread — including when no backlog quota is configured at all.
The metric it feeds is already gated: emission goes through `TopicStats.printTopicStats`, which `NamespaceStatsAggregator` only reaches when `exposeTopicLevelMetricsInPrometheus` is true. Operators who disable topic-level metrics to control cardinality would pay the full cost for a metric they never scrape. The only existing knob is `backlogQuotaCheckEnabled=false`, and even that only stops the periodic checker — the producer-creation path below still triggers the same walk.
Precedent for default-off on expensive per-entity stats: `exposePreciseBacklogInPrometheus`, `exposeSubscriptionBacklogSizeInPrometheus`, `exposeConsumerLevelMetricsInPrometheus`, `exposeManagedCursorMetricsInPrometheus`.
The flag needs to gate the computation here, the `subscriptionOldestPositionInfos` allocation, and the metric emission in `TopicStats` — gating emission alone saves nothing, because the cost lives on the quota-checker path, not the scrape path. Note it would also blank the new `SubscriptionStats.oldestBacklogMessageAgeSeconds` admin field to `-1` when disabled, which seems like the right trade but is worth stating in the config doc.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
Review Comment:
Two separate concerns here.
**Cost.** This is a full, non-short-circuiting walk of every cursor. The pass already called `hasBacklogs()`, which short-circuits on the first backlogged subscription, and then did one estimate (or one `asyncReadEntry` for the single oldest cursor) for the topic. What is new is that every cursor now gets its own `hasBacklog()` call, and every backlogged cursor whose mark-delete moved gets its own estimate or read.
In fairness, the reusable check on line 4023 blunts the steady state: an idle, abandoned backlogged subscription is a cache hit forever after pass 1, so no repeat read. Recurring reads hit cursors that are backlogged *and advancing*.
**Scope.** `managedCursorContainer` is not just user subscriptions. `ManagedLedgerImpl.addCursor` puts durable *and* non-durable cursors into the same container (non-durable ones simply get a null ordering position and stay out of the heap), and the iterator walks the whole map. So this also visits every Reader cursor, `pulsar.repl.*` and `pulsar.dedup`.
For `pulsar.repl.*` and `pulsar.dedup` specifically the work is wasted: `createPersistentSubscriptions()` (line 541) deliberately excludes exactly those two from the `subscriptions` map, so nothing can ever surface their value. (`__compaction` is not in that category — it does become a regular `PersistentSubscription`, so its value is surfaced.) The dedup cursor is the worst case: its mark-delete only advances on snapshots, so on an active topic it near-permanently "has backlog" and also fails the reusable check, meaning a fresh read every pass in precise mode.
The comment on line 4129 ("`ledger.getCursors()` only managed durable cursors") is only true of the oldest-cursor heap, not of this iteration.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -4054,7 +4201,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
future.completeExceptionally(exception);
}
}, null);
- return future;
+ return FutureUtil.waitForAll(List.of(subscriptionOldPositionInfoUpdateFuture, future));
Review Comment:
This branch only runs with `preciseTimeBasedBacklogQuotaCheck=true` (non-default), but for the installs that do enable it, this join puts the per-subscription reads on the producer-creation path.
`ServerCnx` calls `topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age)` on every producer create; with the default `backlogQuotaDefaultRetentionPolicy=producer_request_hold` that reaches `checkTimeBacklogExceeded(true)` -> `updateOldPositionInfo()`. And `checkTimeBacklogExceeded` runs the update *before* testing `backlogQuotaLimitInSecond <= 0`, so it happens even with no time-based quota configured (the default `-1`).
Previously this returned `future` — one read, for the oldest cursor. Now the attach also waits for every per-subscription read. The reads are all issued before anything is awaited, so the added latency is the slowest of N concurrent reads rather than N sequential ones — but with `InflightReadsLimiter` permits shared with consumer traffic, N concurrent cold reads per attach is exactly what turns into queueing (up to `managedLedgerMaxReadsInFlightPermitsAcquireTimeoutMillis`, 60s by default), and reconnect storms multiply it.
Nothing on the quota-enforcement path consumes the per-subscription values, so they should not need to be part of this join at all.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
Review Comment:
A single container-wide version, taken from the *oldest* cursor, is stamped onto every subscription's entry — and `updateSubscriptionResultIfNewer` (line 3986) only replaces on `> 0`. That silently drops legitimately changed values for non-durable cursors.
Non-durable mark-delete advances never bump the container version: `ManagedLedgerImpl.onCursorMarkDeletePositionUpdated` returns early for `!cursor.isDurable()` before reaching `cursors.cursorUpdated`, and `ManagedCursorContainerImpl.cursorUpdated` itself returns at `item.idx == -1` *before* `version = DataVersion.getNextVersion(version)`.
Concrete case: a topic with one idle or abandoned durable subscription plus one backlogged, actively consuming Reader. Nothing calls `add`/`removeCursor`/`cursorUpdated`, so the container version stays at V; the durable cursor's only role is to keep the heap non-empty so the map survives the `clear()` on line 4134. Every pass the Reader's cached entry fails the reusable check because its mark-delete moved, is recomputed with a fresher timestamp, and is then discarded because `compareVersions(V, V) == 0`. The Reader's reported age freezes at its first observation and grows without bound until some durable cursor moves, or any cursor is added or removed on the managed ledger (a reader reconnect is enough, since `add`/`removeCursor` bump the version too).
If the intent is only to order concurrent async completions for the *same* subscription, a per-cursor sequence — or simply comparing the stored mark-delete position — would be more direct than the container version.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
+ activeSubscriptionNames.put(subscriptionName, true);
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ if (markDeletePosition == null || !cursor.hasBacklog(preciseTimeBasedBacklogQuotaCheck)) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ continue;
+ }
+ if (isSubscriptionOldPositionInfoReusable(subscriptionName, markDeletePosition)) {
+ continue;
+ }
+
+ if (preciseTimeBasedBacklogQuotaCheck) {
+ futures.add(updateSubscriptionOldPositionInfoPrecise(
+ subscriptionName, cursor, markDeletePosition, dataVersion));
+ } 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 (Exception e) {
+ log.error()
Review Comment:
This logs at ERROR once per subscription per pass, where the topic-level path produced at most one per topic — and producer-triggered checks can make it more frequent than once per interval.
The same applies to the `readEntryFailed` handler on line 4098, where it matters more: that one is the precise-mode read path, so a persistently unreadable ledger turns into `subscriptions x topics` ERROR lines every check interval. (This site is the estimate branch, which does no BookKeeper read, so it should be much rarer.)
Genuine storage failures can deserve ERROR, so rather than dropping these to DEBUG outright, some aggregation or rate limiting — for example one summary per topic per pass — would be better.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
+ activeSubscriptionNames.put(subscriptionName, true);
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ if (markDeletePosition == null || !cursor.hasBacklog(preciseTimeBasedBacklogQuotaCheck)) {
Review Comment:
Worth noting this runs *before* the reusable-position check on line 4023, so the per-cursor backlog test is paid on every pass even when the cached value would have been reused.
That only matters in precise mode. There, `hasBacklog(true)` delegates to `hasBacklog()`, which can take the cursor read lock and scan `individualDeletedMessages` above the mark-delete position (cheap when there are no ack holes, proportional to the number of deleted ranges when there are). In the default non-precise mode `hasBacklog(false)` is a counter subtraction (falling back to the precise scan only if the counters go negative), so there is nothing meaningful to save there.
Not a straight swap, though: `isSubscriptionOldPositionInfoReusable` dereferences `markDeletePosition`, so it would have to keep the `markDeletePosition == null` guard ahead of it.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
Review Comment:
This decode is wrong for non-durable cursors.
Durable cursors are created with `Codec.encode(subscriptionName)` (line 1162), so `decode` round-trips. Non-durable cursors get the **raw** subscription name — `ledger.newNonDurableCursor(startPosition, subscriptionName, ...)` on line 1250. `Codec.decode` is `URLDecoder.decode`, so:
- A Reader / non-durable subscription named `a+b` is cached under the key `a b`, while `getBestEffortOldestUnacknowledgedMessageAgeSeconds(subName)` on line 3958 looks it up as `a+b` — the value is silently `-1` forever. Same for any name containing a valid `%XX` sequence.
- A name with *malformed* percent-encoding (e.g. `progress-100%`) makes `URLDecoder.decode` throw `IllegalArgumentException`. That happens here, outside the per-cursor `catch` on line 4044, so it escapes `updateOldPositionInfo()` synchronously — which also breaks the "methods returning `CompletableFuture` must not throw synchronously" rule. From `BrokerService.monitorBacklogQuota()` it would propagate out of the `forEachPersistentTopic` lambda; that task is submitted through `scheduleAtFixedRateNonConcurrently`, which (unlike `schedule()`) does not wrap the command in `SafeRunnable`, and `ScheduledFutureTask.run()` only re-queues a periodic task when `runAndReset()` returns true. So an uncaught throw here would stop the broker's backlog quota checker permanently.
`createPersistentSubscriptions()` on line 546 does the same `Codec.decode(cursor.getName())`, but it is safe there: it only runs at topic load, when the container holds just the durable cursors recovered from metadata. This loop runs continuously, after non-durable cursors exist.
Suggest resolving the subscription from `subscriptions` instead of decoding the cursor name — that also skips the replicator/dedup cursors noted above.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java:
##########
@@ -472,6 +472,8 @@ public void backlogsStatsPrecise() throws PulsarAdminException, PulsarClientExce
assertThat(topicStats.getOldestBacklogMessageAgeSeconds())
.isCloseTo(expectedMessageAgeSeconds, within(1L));
assertThat(topicStats.getOldestBacklogMessageSubscriptionName()).isEqualTo(subName2);
+ assertThat(topicStats.getSubscriptions().get(subName2).getOldestBacklogMessageAgeSeconds())
Review Comment:
The new end-to-end assertions are only in `backlogsStatsPrecise`, which sets `preciseTimeBasedBacklogQuotaCheck=true`. The default (non-precise) path — `backlogsStatsNotPrecise` and `backlogsStatsNotPreciseWithNoBacklog` — has no coverage for the new value or metric, and that is the path where the reader-only issue noted on `PersistentTopic.java:4134` lives.
The non-negative assertions here and at line 550 are also only ever made against `subName2`, which is the topic-oldest cursor whose age the topic-level metric already reports; `subName1`'s new value is only asserted at line 576, in the drained state where `subName2` and the topic-level value are `-1` too. So there is no case anywhere with two subscriptions at two *different* non-negative ages — which is the whole point of the PR, so a regression to "every subscription reports the topic-level value" would pass CI today.
Worth adding: two subscriptions at different ages, a non-durable/Reader subscription, and cache cleanup after a subscription is deleted and recreated.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3974,18 +4114,25 @@ public CompletableFuture<Void> updateOldPositionInfo() {
return CompletableFuture.completedFuture(null);
}
- if (!hasBacklogs(brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck())) {
+ boolean preciseTimeBasedBacklogQuotaCheck =
+ brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck();
+ if (!hasBacklogs(preciseTimeBasedBacklogQuotaCheck)) {
log.debug("No backlog. Update old position info is null");
TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.set(this, null);
+ subscriptionOldestPositionInfos.clear();
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> subscriptionOldPositionInfoUpdateFuture =
+ updateSubscriptionOldPositionInfos(managedCursorContainer, preciseTimeBasedBacklogQuotaCheck);
+
// If we have no durable cursor since `ledger.getCursors()` only managed durable cursors
CursorInfo oldestMarkDeleteCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
if (oldestMarkDeleteCursorInfo == null || oldestMarkDeleteCursorInfo.getPosition() == null) {
log.debug("No durable cursor found. Update old position info is null");
TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.set(this, null);
- return CompletableFuture.completedFuture(null);
+ subscriptionOldestPositionInfos.clear();
Review Comment:
This clears everything `updateSubscriptionOldPositionInfos()` just computed on line 4127.
The branch is reached whenever there is no durable cursor — and `getCursorWithOldestPosition()` reads the container *heap*, which non-durable cursors never enter (`ManagedLedgerImpl.addCursor` passes a null ordering position for them). A topic whose subscriptions are all Readers takes this path, because `hasBacklogs()` on line 4119 still passes (Readers do create `PersistentSubscription` entries).
So in the shipped default (non-precise) mode the loop computes every value synchronously and this line throws them all away: `getOldestBacklogMessageAgeSeconds()` is **always `-1`** for those subscriptions. The wasted work there is in-memory only, so the cost is minor — the correctness problem is the point. In precise mode it is worse: the reads were already issued, so the callbacks repopulate the map *after* this `clear()`, making the outcome ordering-dependent and inconsistent with the default mode.
Together with the version issue on line 4013, that leaves non-durable subscriptions wrong in two easily-reachable topologies: `-1` on reader-only topics, and frozen/overstated on a topic whose durable cursors are idle.
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java:
##########
@@ -57,6 +57,16 @@ public interface SubscriptionStats {
/** Get the publish time of the earliest message in the backlog. */
long getEarliestMsgPublishTimeInBacklog();
+ /**
+ * Age of oldest unacknowledged message, as recorded in last backlog quota check interval.
+ * <p>
+ * The age of the oldest unacknowledged (i.e. backlog) message for this subscription, measured by the time elapsed
+ * from its published time, in seconds. This value is recorded every backlog quota check interval, hence it
+ * represents the value seen in the last check.
+ * </p>
+ */
+ long getOldestBacklogMessageAgeSeconds();
Review Comment:
Adding an abstract method to this public admin-api interface is source-breaking for any third-party implementation of `SubscriptionStats`, and it can produce `AbstractMethodError` in a mixed-jar deployment — code compiled against the new interface calling this getter on a third-party implementation compiled against an older one.
I know this interface has consistently added plain abstract methods (PIP-379's draining-hashes getters, PIP-406's `getDispatchThrottled*`), so this follows local precedent and I won't insist. But a `default` returning `-1` costs nothing here, keeps the sentinel semantics, and would make the addition compatible.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -4006,9 +4153,9 @@ public CompletableFuture<Void> updateOldPositionInfo() {
.attr("name", oldestMarkDeleteCursorInfo.getCursor().getName())
.log("Updating cached old position info, " + "since cursor causing it has changed from to");
}
- return CompletableFuture.completedFuture(null);
+ return subscriptionOldPositionInfoUpdateFuture;
}
- if (brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck()) {
+ if (preciseTimeBasedBacklogQuotaCheck) {
Review Comment:
Minor: the loop above already covered the oldest cursor, so this repeats its work — the precise branch reads `getNextValidPosition(oldestMarkDeletePosition)`, the same position `updateSubscriptionOldPositionInfoPrecise` read on line 4066, and the `else` branch repeats `estimatedTimeBasedBacklogQuotaCheck(oldestMarkDeletePosition)`.
Both reads are issued in the same synchronous stretch, so `PendingReadsManager` will normally coalesce them into one storage read and the practical cost is small. Still, the topic-level value could just be derived from the per-subscription result for the oldest cursor's name, which would remove the duplicate work and the extra dependency.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java:
##########
@@ -324,6 +324,9 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st
writeSubscriptionMetric(stream, "pulsar_subscription_back_log_no_delayed",
subsStats.msgBacklogNoDelayed, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel,
customLabelsAndValues);
+ writeSubscriptionMetric(stream, "pulsar_subscription_storage_backlog_age_seconds",
Review Comment:
This is emitted unconditionally for every subscription, so every broker gains one extra time series per subscription per scrape whether or not anyone uses the feature — `exposeTopicLevelMetricsInPrometheus` defaults to `true`.
For a backlogged persistent subscription the value is real. But it is a constant `-1` for every subscription with no backlog, and for every subscription on a non-persistent topic: `aggregateTopicStats` does assign the field, it just copies the `-1` default of `SubscriptionStatsImpl.oldestBacklogMessageAgeSeconds`, because only `PersistentSubscription.getStats` ever populates the source.
Worth tying to the proposed config flag, and arguably skipping the sample entirely when the value is `-1`, since it carries no signal.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
+ activeSubscriptionNames.put(subscriptionName, true);
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ if (markDeletePosition == null || !cursor.hasBacklog(preciseTimeBasedBacklogQuotaCheck)) {
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ continue;
+ }
+ if (isSubscriptionOldPositionInfoReusable(subscriptionName, markDeletePosition)) {
+ continue;
+ }
+
+ if (preciseTimeBasedBacklogQuotaCheck) {
+ futures.add(updateSubscriptionOldPositionInfoPrecise(
+ subscriptionName, cursor, markDeletePosition, dataVersion));
+ } 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 (Exception e) {
+ log.error()
+ .attr("subscriptionName", subscriptionName)
+ .exceptionMessage(e)
+ .log("Error updating subscription old position info");
+ clearSubscriptionResultIfNewer(subscriptionName, dataVersion);
+ }
+ }
+ }
+
+ subscriptionOldestPositionInfos.keySet().removeIf(subscriptionName ->
Review Comment:
This cleanup runs synchronously, before the precise-mode futures launched on line 4028 have settled, and `updateSubscriptionResultIfNewer` inserts unconditionally when the key is absent (line 3984 — the version guard is bypassed on insert). Passes can overlap: `monitorBacklogQuota` does not await its fan-out, and producer-creation checks call `updateOldPositionInfo()` concurrently.
So: pass A launches a read, the subscription is unsubscribed, pass B removes the key, the subscription is recreated with the same name, and pass A's callback then inserts the *dead* cursor's publish timestamp against the fresh subscription. It self-heals within about an interval, but the stats are contradictory until then.
Separately, `removeSubscription()` never clears `subscriptionOldestPositionInfos`, so same-name recreation can inherit stale data with no race involved at all.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3952,6 +3954,16 @@ public long getBestEffortOldestUnacknowledgedMessageAgeSeconds() {
}
}
+ long getBestEffortOldestUnacknowledgedMessageAgeSeconds(String subscriptionName) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ if (positionInfo == null) {
+ return -1;
+ } else {
+ return TimeUnit.MILLISECONDS.toSeconds(
+ Clock.systemUTC().millis() - positionInfo.getPositionPublishTimestampInMillis());
Review Comment:
`AbstractTopic` already holds an injectable `clock` (`brokerService.getClock()`), and `estimatedTimeBasedBacklogQuotaCheck` uses it. Using `Clock.systemUTC()` here makes the new value impossible to test with a controlled clock. The topic-level accessor just above has the same issue, but it would be good not to propagate it into new code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#issuecomment-5423282207
> It may be worth adding top-level exception handling around refreshSubscriptionBacklogAge().join(). Since scheduleAtFixedRateNonConcurrently does not wrap the task with SafeRunnable, an unexpected unchecked exception could prevent subsequent executions from being requeued. Adding a guard here would make the periodic refresh more resilient.
thanks 🌸
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3793896261
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
Review Comment:
addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #26313:
URL: https://github.com/apache/pulsar/pull/26313
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3793926467
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
+ String subscriptionName = Codec.decode(cursor.getName());
+ activeSubscriptionNames.put(subscriptionName, true);
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ if (markDeletePosition == null || !cursor.hasBacklog(preciseTimeBasedBacklogQuotaCheck)) {
Review Comment:
addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3793906017
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3966,6 +3978,134 @@ private void updateResultIfNewer(OldestPositionInfo updatedResult) {
}
+ private void updateSubscriptionResultIfNewer(String subscriptionName, OldestPositionInfo updatedResult) {
+ subscriptionOldestPositionInfos.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) {
+ subscriptionOldestPositionInfos.computeIfPresent(subscriptionName,
+ (__, existingResult) ->
+ ManagedCursorContainer.DataVersion.compareVersions(
+ dataVersion, existingResult.getDataVersion()) >= 0 ? null : existingResult);
+ }
+
+ private boolean isSubscriptionOldPositionInfoReusable(String subscriptionName, Position markDeletePosition) {
+ OldestPositionInfo positionInfo = subscriptionOldestPositionInfos.get(subscriptionName);
+ return positionInfo != null
+ && markDeletePosition.compareTo(positionInfo.getOldestCursorMarkDeletePosition()) == 0
+ && markDeletePosition.compareTo(ledger.getFirstPosition()) >= 0;
+ }
+
+ private CompletableFuture<Void> updateSubscriptionOldPositionInfos(ManagedCursorContainer managedCursorContainer,
+ boolean preciseTimeBasedBacklogQuotaCheck) {
+ Map<String, Boolean> activeSubscriptionNames = new HashMap<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ CursorInfo oldestCursorInfo = managedCursorContainer.getCursorWithOldestPosition();
+ long dataVersion = oldestCursorInfo == null ? 0 : oldestCursorInfo.getVersion();
+
+ for (ManagedCursor cursor : managedCursorContainer) {
Review Comment:
improved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Add pulsar_subscription_storage_backlog_age_seconds metric [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #26313:
URL: https://github.com/apache/pulsar/pull/26313#discussion_r3764248989
##########
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);
Review Comment:
This adds subscription-level backlog age computation (via per-subscription cached oldest-position info), but there doesn't appear to be an end-to-end broker test asserting that the per-subscription value is populated/updated after backlog quota checks (similar to the existing topic-level assertions). Adding a broker test that produces messages, triggers backlogQuotaCheck, and asserts getBestEffortOldestUnacknowledgedMessageAgeSeconds(subName) > 0 (and/or the Prometheus metric) would help prevent regressions.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3952,6 +3954,16 @@ public long getBestEffortOldestUnacknowledgedMessageAgeSeconds() {
}
}
+ public long getBestEffortOldestUnacknowledgedMessageAgeSeconds(String subscriptionName) {
Review Comment:
The new overload appears to be an internal helper used only by PersistentSubscription. Consider reducing its visibility (package-private) to avoid expanding PersistentTopic's public surface area unnecessarily and to prevent accidental misuse (e.g., passing an encoded cursor name instead of a decoded subscription name).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org