[feat][broker] Implement topic-level delayed delivery tracking with in-memory manager

[Denovo1998]

Fixes #xyz

Main Issue: apache#24600

PIP: apache#24928

Motivation

The primary motivation for this proposal is to address the high memory consumption caused by the current per-subscription delayed message tracking mechanism. For topics with hundreds or thousands of subscriptions, the memory footprint for delayed messages becomes prohibitively large. Each delayed message's position is duplicated across every subscription's tracker, leading to a memory usage pattern of O(num_delayed_messages * num_subscriptions).

This excessive memory usage can cause:

Increased memory pressure on Pulsar brokers.
More frequent and longer Garbage Collection (GC) pauses, impacting broker performance.
Potential OutOfMemoryErrors, leading to broker instability.
Limited scalability for use cases that rely on many subscriptions per topic, such as IoT or large-scale microservices with shared subscriptions.
By optimizing the delayed message tracking to be more memory-efficient, we can enhance broker stability and scalability, allowing Pulsar to better support these critical use cases.

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: apache#24927

[Copilot]

Pull Request Overview

This PR introduces a topic-level delayed delivery tracker manager for in-memory delayed message tracking, significantly optimizing memory usage in multi-subscription scenarios. Instead of maintaining separate delayed message indexes for each subscription, a single global index is shared across all subscriptions of a topic, with per-subscription views filtering based on mark-delete positions.

Key changes:

• Introduces TopicDelayedDeliveryTrackerManager interface and InMemoryTopicDelayedDeliveryTrackerManager implementation with a shared global delayed message index per topic
• Adds InMemoryTopicDelayedDeliveryTrackerView as subscription-specific views that implement the DelayedDeliveryTracker interface
• Updates InMemoryDelayedDeliveryTrackerFactory to cache and reuse topic-level managers across subscriptions
• Implements event-driven mark-delete position updates via markDeletePositionMoveForward() callback in dispatcher implementations
• Adds comprehensive unit tests for the new topic-level tracking mechanism

Reviewed Changes

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

Show a summary per file

File	Description
TopicDelayedDeliveryTrackerManager.java	New interface defining the contract for topic-level delayed delivery tracking
InMemoryTopicDelayedDeliveryTrackerManager.java	Core implementation managing shared delayed message index with subscription contexts and per-bucket locking
InMemoryTopicDelayedDeliveryTrackerView.java	Per-subscription view wrapper delegating to the topic-level manager
InMemoryDelayedDeliveryTrackerFactory.java	Updated to instantiate and cache topic-level managers instead of per-subscription trackers
PersistentDispatcherMultipleConsumers.java	Implements mark-delete position callback for in-memory topic tracker views
PersistentDispatcherMultipleConsumersClassic.java	Implements mark-delete position callback for in-memory topic tracker views
DelayedDeliveryTrackerFactoryTest.java	Updates test assertions to reference the new view class name
InMemoryTopicDeliveryTrackerTest.java	Comprehensive test suite covering concurrent operations, pruning, timer scheduling, and multi-subscription scenarios
InMemoryDelayedDeliveryTrackerFactoryUnitTest.java	Tests factory-level behavior including manager caching and lifecycle

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The comment indicates a known limitation where the returned count includes messages that may not be visible to this specific subscription (filtered by mark-delete position). This discrepancy between the method's semantic meaning and its actual behavior should be clearly documented in the method's JavaDoc to avoid confusion for API consumers.

[Copilot]

Similar to the view class, this method's name suggests it returns subscription-specific visible messages, but it returns the global count. The comment acknowledges this limitation, but the method name getNumberOfVisibleDelayedMessagesForSub is misleading. Consider either renaming to clarify it returns an approximation or implementing accurate per-subscription counting.

Suggested change

	private long getNumberOfVisibleDelayedMessagesForSub(SubContext subContext) {
	private long getApproximateNumberOfVisibleDelayedMessagesForSub(SubContext subContext) {

[Denovo1998]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Missing reschedule leaves later delayed messages undelivered

After the timer fires, run() clears currentTimeoutTarget and timeout but never schedules a new timeout for the next delayed bucket. If a topic has multiple delayed timestamps queued (e.g. messages at T1 and T2>T1) and no new messages are added after T1, the first tick triggers delivery but the remaining messages never become visible because the manager will not call readMoreEntriesAsync again when time reaches T2. The previous per-subscription tracker re‑armed its timer in getScheduledMessages; this implementation does not, so delayed messages can be stuck indefinitely until another message is added or tick time changes.

Useful? React with 👍 / 👎.

[codecov]

Codecov Report

❌ Patch coverage is 78.77358% with 90 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.31%. Comparing base (ee33c99) to head (4aada39).
⚠️ Report is 11 commits behind head on master.

Files with missing lines	Patch %	Lines
...ed/InMemoryTopicDelayedDeliveryTrackerManager.java	82.14%	27 Missing and 28 partials ⚠️
...ed/InMemoryTopicDelayedDeliveryTrackerFactory.java	50.84%	27 Missing and 2 partials ⚠️
...AbstractPersistentDispatcherMultipleConsumers.java	60.00%	3 Missing and 1 partial ⚠️
...r/delayed/InMemoryTopicDelayedDeliveryTracker.java	94.28%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master      #16      +/-   ##
============================================
+ Coverage     74.28%   74.31%   +0.03%     
- Complexity     2492    33588   +31096     
============================================
  Files          1913     1916       +3     
  Lines        149280   149738     +458     
  Branches      17324    17401      +77     
============================================
+ Hits         110887   111277     +390     
- Misses        29565    29599      +34     
- Partials       8828     8862      +34     

Flag	Coverage Δ
inttests	26.27% <3.30%> (-0.39%)	⬇️
systests	22.69% <2.83%> (-0.19%)	⬇️
unittests	73.84% <78.77%> (+0.05%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...org/apache/pulsar/broker/ServiceConfiguration.java	98.21% <100.00%> (+<0.01%)	⬆️
...delayed/InMemoryDelayedDeliveryTrackerFactory.java	95.23% <100.00%> (ø)
...sistent/PersistentDispatcherMultipleConsumers.java	75.35% <100.00%> (+0.10%)	⬆️
.../PersistentDispatcherMultipleConsumersClassic.java	48.27% <100.00%> (+0.56%)	⬆️
...ersistentStickyKeyDispatcherMultipleConsumers.java	85.30% <100.00%> (+0.04%)	⬆️
...ntStickyKeyDispatcherMultipleConsumersClassic.java	87.19% <100.00%> (+0.04%)	⬆️
...r/delayed/InMemoryTopicDelayedDeliveryTracker.java	94.28% <94.28%> (ø)
...AbstractPersistentDispatcherMultipleConsumers.java	66.66% <60.00%> (-33.34%)	⬇️
...ed/InMemoryTopicDelayedDeliveryTrackerFactory.java	50.84% <50.84%> (ø)
...ed/InMemoryTopicDelayedDeliveryTrackerManager.java	82.14% <82.14%> (ø)

... and 86 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.