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

[Denovo1998]

Fixes #xyz

Main Issue: #24600

PIP: #xyz

Motivation

Modifications

1. TopicDelayedDeliveryTrackerManager Interface - Created the unified interface for topic-level delayed delivery management
2. InMemoryTopicDelayedDeliveryTrackerManager - Implemented the topic-level manager with:

• Global shared delayed message index using Long2ObjectSortedMap<Long2ObjectSortedMap<Roaring64Bitmap>>
• Per-subscription context management
• Timer-based delayed delivery scheduling
• Memory optimization through timestamp precision trimming
• Pruning based on minimum mark delete position across subscriptions

3. InMemoryTopicDelayedDeliveryTrackerView - Implemented per-subscription view objects that:

• Forward all operations to the topic-level manager
• Maintain compatibility with existing DelayedDeliveryTracker interface

4. InMemoryDelayedDeliveryTrackerFactory - Updated to use the new topic-level manager architecture

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: #24927

[Copilot]

Pull Request Overview

This PR introduces a topic-level delayed delivery tracking system to optimize memory usage in multi-subscription scenarios. Instead of maintaining separate delayed message indices per subscription, the new implementation uses a shared global index at the topic level with per-subscription views.

Key changes:

• Introduces TopicDelayedDeliveryTrackerManager interface and InMemoryTopicDelayedDeliveryTrackerManager implementation for shared topic-level delayed message tracking
• Adds InMemoryTopicDelayedDeliveryTrackerView as a per-subscription adapter that delegates to the topic-level manager
• Updates InMemoryDelayedDeliveryTrackerFactory to create and cache topic-level managers and return subscription views
• Updates test assertions to expect the new InMemoryTopicDelayedDeliveryTrackerView type instead of InMemoryDelayedDeliveryTracker

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
TopicDelayedDeliveryTrackerManager.java	New interface defining the contract for topic-level delayed delivery tracking management
InMemoryTopicDelayedDeliveryTrackerManager.java	New implementation of topic-level manager with shared delayed message index and per-subscription context tracking
InMemoryTopicDelayedDeliveryTrackerView.java	New per-subscription view adapter that implements DelayedDeliveryTracker and delegates to the topic-level manager
InMemoryDelayedDeliveryTrackerFactory.java	Updated to create topic-level managers and cache them, returning subscription views instead of individual trackers
DelayedDeliveryTrackerFactoryTest.java	Updated test assertions to expect InMemoryTopicDelayedDeliveryTrackerView instances

---

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

[Copilot]

The topicManagers map could accumulate stale entries for topics that have been unloaded. When a topic is unloaded and all subscriptions are closed, the manager should be removed from the map. Consider adding cleanup logic in InMemoryTopicDelayedDeliveryTrackerManager.unregister() that notifies the factory to remove the manager from the cache, or add a callback mechanism for the manager to remove itself when it closes.

[Copilot]

The updateMarkDeletePosition method is public but not part of the DelayedDeliveryTracker interface. This creates an API inconsistency where callers need to cast to the concrete type to call this method. Consider either adding this method to the interface or finding an alternative design pattern (such as having the dispatcher directly notify the manager) to avoid breaking the interface abstraction.

[Copilot]

Calling close() within unregister() while already holding the lock could lead to maintenance issues. The close() method also acquires the same lock (line 180), which is reentrant but adds complexity. Consider either documenting this self-closing behavior clearly or extracting the cleanup logic to avoid nested locking patterns.

Suggested change

	synchronized (this) {
	subscriptionContexts.remove(subscriptionName);
	// If no more subscriptions, close the manager
	if (subscriptionContexts.isEmpty() && delayedMessageMap.isEmpty()) {
	close();
	}
	}
	boolean shouldClose = false;
	synchronized (this) {
	subscriptionContexts.remove(subscriptionName);
	// If no more subscriptions, close the manager
	if (subscriptionContexts.isEmpty() && delayedMessageMap.isEmpty()) {
	shouldClose = true;
	}
	}
	if (shouldClose) {
	close();
	}

[Copilot]

Logging a warning every time onTickTimeUpdated is called could spam logs if this method is invoked frequently. Consider either implementing the functionality, using a rate-limited logger, or throwing an UnsupportedOperationException to clearly indicate this is not supported rather than silently logging warnings.

Suggested change

	// For now, tick time updates are not supported after initialization
	// This could be enhanced to update all subscription contexts
	log.warn("Tick time updates are not currently supported for topic-level delayed delivery managers");
	throw new UnsupportedOperationException("Tick time updates are not currently supported for topic-level delayed delivery managers");

[Copilot]

The createOrGetView method always creates a new InMemoryTopicDelayedDeliveryTrackerView instance even when the SubContext already exists. This means multiple view instances could be created for the same subscription, each holding a reference to the same SubContext. When multiple views are closed, the unregister method will remove the subscription context prematurely, breaking other view instances. The method should track and return existing view instances or document that only one view per subscription should be active at a time.

[Copilot]

Double space between = and Mockito.spy(tracker). Should be a single space for consistency with code formatting standards.

Suggested change

	tracker = Mockito.spy(tracker);
	tracker = Mockito.spy(tracker);

[lhotari]

It would be worthwhile to create a PIP to cover these changes. In this case the reference is currently to #24600 as the motivation. However, there's no clear problem description or goals what is done and why. For reviewers, it's most necessary to understand the problem that the PR intends to solve. That's one role of a PIP document.
For example, PIP-430 covered broker cache refactorings and improvements. With an LLM, you can generate most of the PIP document by providing relevant details as context documents to your prompt.
The process is somewhat similar to the one described for PR descriptions with LLMs.
Context:

• GitHub issues as printed as PDFs from the browser or copy-pasted to text files
• PR description printed as PDF from the browser
• implementation PR patch (append .patch to PR url to get the patch file, download and save to .patch.txt file)
• pip/README.md
• pip/TEMPLATE.md
• pip/pip-379.md and pip/pip-430.md as examples
  The prompt could be such as

I'm writing "PIP-XXX: Optimized in-memory delayed delivery tracker".
Please take a look at the instructions and samples of PIP-379 and PIP-430 and create a pip-XXX.md document that it's a professional Apache Pulsar Improvement Proposal that I could send out for discussion in the Apache Pulsar development community. The expected audience is IT architects and Software Engineers specialized in data streaming.
Return the document as a downloadable markdown file called pip-XXX.md

LLM will generate 80% of the PIP document if you already have written a lot of the background material in other discussions. This solution requires a LLM model with sufficient context window size. If the window size is not sufficient, you can use text files instead of printed PDFs for the existing issues and background context discussions.

[Denovo1998]

@lhotari I‘m sorry that I mentioned that I chose the wrong repo. Originally, I was going to mention my own repo. Indeed, this is a big change that requires a PIP. This is only a preliminary implementation. I have found a lot of problems. There are more changes to be made.

[Denovo1998]

Yes. I have written a prompt before, which can generate documents in the correct format.

https://gist.github.com/Denovo1998/163e55b3a612873364a00cf0df5a1b95