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

[Denovo1998]

Fixes #24603

Main Issue: #24600

PIP: #xyz

Motivation

The current concurrent control model of BucketDelayedDeliveryTracker has serious performance bottlenecks and thread safety issues. It mainly relies on coarse-grained synchronized keywords to protect its internal state, which leads to several key problems:

1. High lock contention

`All operations, whether simple read-only checks (like containsMessage) or complex I/O-intensive operations (like creating a Bucket snapshot), contend for the same object lock. This leads to operations being unnecessarily serialized, limiting overall throughput.

2. Read-write operations mutually block

• Write operations block read operations: An executing addMessage or getScheduledMessages will block all containsMessage and nextDeliveryTime and other read-only checks.
• Read-write mixed operations block all other operations: for example, the getScheduledMessages method, which appears to be reading messages but actually includes multiple write operations such as modifying the queue, updating the bitmap, and removing mappings. When executed within a synchronized block, it holds the lock for a long time, blocking the addition of all new messages.

3. I/O operations block the critical path
   Within the synchronized block of addMessage, the creation and persistence of Bucket snapshots are executed synchronously. This is a time-consuming I/O operation that blocks all subsequent message addition and read requests until the snapshot is completed, significantly increasing the message publishing delay.

This PR is the first step in optimizing and enhancing the BucketDelayedDeliveryTracker, aiming to address the above performance bottlenecks by introducing more refined concurrency control mechanisms and asynchronizing I/O operations.

Modifications

This change is mainly focused on the addMessage method, because it is one of the most critical bottlenecks under high throughput scenarios. Subsequent PRs will continue to optimize other methods based on this foundation, such as getScheduledMessages.

1. Introducing fine-grained locks (ReentrantReadWriteLock):

• Removed StampedLock and the synchronized keyword on multiple methods, unifying the use of ReentrantReadWriteLock as the core concurrency control mechanism.

• This implements read-write separation:

  • Read operations (containsMessage, nextDeliveryTime) now use a read lock, allowing multiple threads to execute concurrently without blocking each other.

  • Write operations (addMessage, asyncMergeBucketSnapshot) use a write lock, ensuring atomicity and consistency of data modifications.

• addMessage internally adopts an optimized pattern of "read lock check -> write lock modification" to reduce the holding time of the write lock.

2. Asynchronous Bucket Snapshot Creation:

• Introduced a dedicated single-threaded ExecutorService (bucketSnapshotExecutor) to handle time-consuming Bucket snapshot persistence tasks.

• Refactored the snapshot creation logic in addMessage:

  • When a snapshot needs to be created, no longer wait synchronously. Instead, mark the current MutableBucket as bucketBeingSealed.

  • Immediately create a new lastMutableBucket to handle subsequent incoming messages, ensuring that addMessage calls can return quickly.

  • Submit the actual snapshot creation and persistence task to the background bucketSnapshotExecutor for asynchronous execution.

• This change removes I/O operations from the critical path of message publishing, greatly reducing publish latency and increasing throughput.

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: Denovo1998#10

[Denovo1998]

@codelipenghui @lhotari @coderzc @Apurva007
How does this copy on write method look, is my direction correct?
There is still much work to be done, and I will convert this PR to a draft later.

[lhotari]

@codelipenghui @lhotari @coderzc @Apurva007
How does this copy on write method look, is my direction correct?
There is still much work to be done, and I will convert this PR to a draft later.

Without benchmarking, it will be hard to validate assumptions. I'd suggest adding JMH benchmarks so that performance could be compared.

JMH benchmarks can be added to microbench module: https://github.com/apache/pulsar/tree/master/microbench

[Denovo1998]

OK. I will add JMH benchmarks first.

[Denovo1998]

bump, this PR can be reviewed now.
@lhotari @codelipenghui @coderzc @Apurva007 @thetumbled @dao-jun @BewareMyPower PTAL.