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

[Copilot]

Main Issue:

Motivation

The BucketDelayedDeliveryTracker has thread safety issues with its bucket sealing and message routing logic. The current implementation has race conditions when:

• A bucket is being sealed while new messages arrive for that bucket's ledger range
• The bucketBeingSealed field is accessed without proper synchronization in cleanup paths
• hasMessageAvailable() unnecessarily acquires the write lock when it only reads state

Modifications

Thread-safe bucket sealing with explicit handoff:

• Pass bucketToSeal explicitly to createBucketSnapshotAsync() instead of reading from bucketBeingSealed field after the lock is released
• Messages arriving for a bucket being sealed now route to sharedBucketPriorityQueue and track via current lastMutableBucket bitmap, avoiding writes to the detached bucket

Proper lock acquisition for state cleanup:

• Added clearBucketBeingSealed() and clearBucketBeingSealedUnsafe() helper methods for consistent cleanup under write lock
• All exception paths now properly clear bucketBeingSealed under lock

Read-only operations use read lock:

• Changed hasMessageAvailable() from write lock to read lock since it only reads state

Improved executor shutdown:

• Added proper awaitTermination() for bucketSnapshotExecutor in close()
• Named thread factory includes context name for better debugging

Documentation:

• Added detailed Javadoc explaining routing rules and concurrency invariants

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as BucketDelayedDeliveryTrackerThreadSafetyTest.

Does this pull request potentially affect one of the following parts:

• The threading model

Documentation

• doc-not-needed

Matching PR in forked repository

PR in forked repository: apache#24739

---

💬 We'd love your input! Share your thoughts on Copilot coding agent in our 2 minute survey.