[refactor][broker] Decouple delayed delivery trackers from dispatcher by Denovo1998 · Pull Request #25384 · apache/pulsar

Denovo1998 · 2026-03-22T13:29:13Z

Main Issue: #24600

Related PR: #24739

Motivation

PR #24739 currently mixes correctness and threading changes with benchmark and testability refactors, which makes the review difficult to scope. This preparatory PR extracts the delayed delivery tracker context that is needed to construct trackers without a real dispatcher, so the benchmark and test scaffolding can be reviewed separately. The goal is to keep the existing production behavior intact while making the tracker setup reusable from microbench code.

Modifications

introduce DelayedDeliveryContext plus dispatcher-backed and no-op implementations for delayed delivery trackers
add context-based and name+cursor-based constructors to the delayed delivery trackers, while keeping the existing dispatcher-based production entry points intact
update the delayed delivery tracker factories to expose benchmark/test construction helpers without changing the production factory flow
make MockManagedCursor public for microbench usage and switch the JMH benchmark setup to the new context-based tracker construction
add MockBucketSnapshotStorage, add the new BucketDelayedDeliveryTrackerBenchmark, and remove the old simple benchmark

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

Documentation

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

Matching PR in forked repository

PR in forked repository: Denovo1998#24

…text handling in delayed delivery trackers

… neutral

github-actions · 2026-03-22T13:29:41Z

@Denovo1998 Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

Denovo1998 · 2026-03-22T13:43:04Z

@codelipenghui @lhotari @merlimat @coderzc @BewareMyPower @dao-jun
PTAL!

Copilot

Pull request overview

Refactors delayed delivery trackers to depend on a lightweight DelayedDeliveryContext abstraction instead of directly depending on the dispatcher, enabling tracker construction in benchmarks/tests without a real dispatcher while keeping the production entry points intact.

Changes:

Introduce DelayedDeliveryContext plus dispatcher-backed (DispatcherDelayedDeliveryContext) and no-op (NoopDelayedDeliveryContext) implementations.
Add context-based (and name+cursor-based) constructors/creation helpers for InMemoryDelayedDeliveryTracker and BucketDelayedDeliveryTracker, while preserving existing dispatcher-based construction.
Update microbench scaffolding: remove the old simple benchmark, add a new BucketDelayedDeliveryTrackerBenchmark, add MockBucketSnapshotStorage, and make MockManagedCursor public for reuse.

Reviewed changes

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

Show a summary per file

File	Description
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryContext.java	New context interface to decouple trackers from dispatcher.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DispatcherDelayedDeliveryContext.java	Production context backed by dispatcher operations.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/NoopDelayedDeliveryContext.java	Test/bench-friendly context implementation.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java	Base tracker now uses `DelayedDeliveryContext` and a context-provided trigger lock.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java	Adds context/name+cursor constructors; replaces dispatcher references with context.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java	Uses dispatcher-backed context internally; adds test/bench helper constructor path.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java	Uses dispatcher-backed context internally; adds test/bench helper constructor path.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java	Adds context/name+cursor constructors; replaces dispatcher references with context.
microbench/src/main/java/org/apache/pulsar/broker/package-info.java	Adds package-level Javadoc (currently targets a higher-level package).
microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/MockBucketSnapshotStorage.java	Adds a microbench snapshot storage stub.
microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java	New benchmark using context-based tracker construction.
microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerSimpleBenchmark.java	Removes old simplified benchmark.
microbench/src/main/java/org/apache/bookkeeper/mledger/impl/MockManagedCursor.java	Makes cursor mock public and factory method public for benchmark reuse.

microbench/src/main/java/org/apache/pulsar/broker/package-info.java

...main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java

microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/MockBucketSnapshotStorage.java

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/NoopDelayedDeliveryContext.java

# Conflicts: # microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerSimpleBenchmark.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

Denovo1998 · 2026-04-03T08:29:51Z

@nodece
All reviews have been processed.

...ar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java

...er/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java

...ar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java

...oker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java

Denovo1998 · 2026-04-03T09:18:22Z

@nodece All done.

Copilot

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Copilot · 2026-04-03T10:00:25Z

...main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java

+    @Benchmark
+    public boolean benchmarkMixedOperations() {
+        String[] parts = readWriteRatio.split("_");
+        int readPercentage = Integer.parseInt(parts[0]);
+
+        if (ThreadLocalRandom.current().nextInt(100) < readPercentage) {


benchmarkMixedOperations() parses readWriteRatio (split + Integer.parseInt) on every benchmark invocation. That parsing overhead will be included in the measured throughput and can dominate at high rates. Consider parsing the ratio once in @Setup (e.g., store an int readPercentage field) and using that field in the benchmark methods.

Copilot · 2026-04-03T10:00:25Z

...main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java

+        // Preload messages to create realistic test conditions
+        long baseTime = System.currentTimeMillis() + 10000; // Future delivery
+        for (int i = 1; i <= initialMessages; i++) {
+            tracker.addMessage(i, i, baseTime + i * 1000);


The benchmark uses System.currentTimeMillis()-based delivery timestamps (e.g., preloadMessages() and performWriteOperation()), which can cause the tracker's Netty timer to start firing during the measurement window (and introduce noisy, non-deterministic background work). Consider using delivery times far beyond the trial duration (or a fixed base timestamp) so timeouts don't trigger while measuring tracker operations.

Suggested change

// Preload messages to create realistic test conditions

long baseTime = System.currentTimeMillis() + 10000; // Future delivery

for (int i = 1; i <= initialMessages; i++) {

tracker.addMessage(i, i, baseTime + i * 1000);

// Preload messages to create realistic test conditions while keeping

// delivery timestamps far beyond the benchmark trial duration so the

// tracker's timer does not start firing during measurement.

long baseTime = 4102444800000L; // 2100-01-01T00:00:00Z

for (int i = 1; i <= initialMessages; i++) {

tracker.addMessage(i, i, baseTime + i * 1000L);

Copilot · 2026-04-03T10:00:25Z

...main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java

+    private boolean performWriteOperation() {
+        long deliverAt = System.currentTimeMillis() + ThreadLocalRandom.current().nextLong(5000, 30000);
+        return addMessageSequential(deliverAt, 1000);
+    }


performWriteOperation() calls System.currentTimeMillis() on the hot path. Besides adding unrelated overhead, it also makes the scheduled timeout firing time depend on wall-clock timing/jitter. Consider deriving deliverAt from a precomputed base timestamp (set up in @Setup) to reduce noise and improve benchmark repeatability.

dao-jun

LGTM

dao-jun · 2026-04-03T18:31:02Z

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

+        this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, Clock.systemUTC(),
+                isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket,
+                timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets);
+    }
+
+    public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
+                                        Timer timer, long tickTimeMillis, Clock clock,
+                                        boolean isDelayedDeliveryDeliverAtTimeStrict,
+                                        BucketSnapshotStorage bucketSnapshotStorage,
+                                        long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis,
+                                        int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets)
+            throws RecoverDelayedDeliveryTrackerException {
+        this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, clock,
+                isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket,
+                timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets);
+    }
+
+    public BucketDelayedDeliveryTracker(String dispatcherName, ManagedCursor cursor,
+                                        Timer timer, long tickTimeMillis,
+                                        boolean isDelayedDeliveryDeliverAtTimeStrict,
+                                        BucketSnapshotStorage bucketSnapshotStorage,
+                                        long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis,
+                                        int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets)
+            throws RecoverDelayedDeliveryTrackerException {
+        this(new NoopDelayedDeliveryContext(dispatcherName, cursor), timer, tickTimeMillis, Clock.systemUTC(),
+                isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket,
+                timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets);
+    }
+
+    public BucketDelayedDeliveryTracker(DelayedDeliveryContext context,
+                                        Timer timer, long tickTimeMillis,
+                                        boolean isDelayedDeliveryDeliverAtTimeStrict,
+                                        BucketSnapshotStorage bucketSnapshotStorage,
+                                        long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis,
+                                        int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets)
+            throws RecoverDelayedDeliveryTrackerException {
+        this(context, timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict,


Too many constructors

Denovo1998 added 2 commits December 5, 2025 20:17

[refactor][delayed] Introduce DelayedDeliveryContext for improved con…

8168546

…text handling in delayed delivery trackers

[refactor][delayed] Keep delayed-delivery context extraction behavior…

583987d

… neutral

Denovo1998 mentioned this pull request Mar 22, 2026

[improve][broker] Optimize fine-grained concurrency control for BucketDelayedDeliveryTracker #24739

Draft

15 tasks

github-actions bot added doc-label-missing doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Mar 22, 2026

Denovo1998 mentioned this pull request Mar 22, 2026

[refactor][broker] Decouple delayed delivery trackers from dispatcher Denovo1998/pulsar#24

Open

5 tasks

nodece requested a review from Copilot April 3, 2026 03:06

Copilot started reviewing on behalf of nodece April 3, 2026 03:07 View session

Copilot AI reviewed Apr 3, 2026

View reviewed changes

Denovo1998 added 2 commits April 3, 2026 16:18

[fix][broker] Address prep PR review feedback

3047521

chore(benchmark): remove unused simple benchmark file

cedae89

nodece requested changes Apr 3, 2026

View reviewed changes

refactor(delayed-delivery): remove unused test constructors

3901a76

Denovo1998 requested a review from nodece April 3, 2026 09:18

refactor(delayed-delivery): Add @VisibleForTesting annotation

2657b81

nodece approved these changes Apr 3, 2026

View reviewed changes

nodece requested review from Copilot and lhotari April 3, 2026 09:54

Copilot started reviewing on behalf of nodece April 3, 2026 09:55 View session

Copilot AI reviewed Apr 3, 2026

View reviewed changes

test(benchmark): fix delayed delivery benchmark to prevent timer firing

577d26d

dao-jun approved these changes Apr 3, 2026

View reviewed changes

dao-jun assigned Denovo1998 Apr 3, 2026

dao-jun added the area/broker label Apr 3, 2026

dao-jun added ready-to-test type/refactor Code or documentation refactors. e.g. refactor code structure or methods to improve code readability release/4.2.1 release/4.1.4 release/4.0.10 labels Apr 3, 2026

Conversation

Denovo1998 commented Mar 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Verifying this change

Documentation

Matching PR in forked repository

Uh oh!

github-actions bot commented Mar 22, 2026

Uh oh!

Denovo1998 commented Mar 22, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Denovo1998 commented Apr 3, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Denovo1998 commented Apr 3, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

dao-jun left a comment

Choose a reason for hiding this comment

Uh oh!

dao-jun Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Denovo1998 commented Mar 22, 2026 •

edited

Loading