[fix][broker] Remove lock contention in delayed delivery stats read paths

[nodece]

Motivation

Under a large delayed-delivery workload (~500M delayed messages), jstack analysis
shows the dispatcher lock is held while sealing delayed-delivery buckets.

A broker worker thread spends significant CPU time in:

TripleLongPriorityQueue.siftDown()
  -> pop()
  -> createImmutableBucketAndAsyncPersistent()

while holding both:

PersistentDispatcherMultipleConsumers
BucketDelayedDeliveryTracker

As a result, stats collection threads (Prometheus, admin APIs, getStatsAsync)
are blocked waiting for the dispatcher monitor.

The same pattern affects InMemoryDelayedDeliveryTracker, where
getBufferMemoryUsage() iterates the entire TreeMap<Long, TreeMap<Long, Roaring64Bitmap>>
while holding the dispatcher lock.

Modifications

1. BucketDelayedDeliveryTracker — lock-free bucket stats via AtomicLong counters

Introduce two AtomicLong counters (bucketsCount and totalSnapshotLengthBytes)
maintained alongside the existing TreeRangeMap<Long, ImmutableBucket>:

• Add private helper methods putBucket(), removeBucket(), and updateBucketSnapshotLength()
  that wrap TreeRangeMap operations and update counters atomically
• putBucket() calculates removed bucket lengths before insertion (since TreeRangeMap.put()
  silently removes/splits overlapping entries) and updates counters by delta
• removeBucket() decrements counters only when removal succeeds
• updateBucketSnapshotLength() updates counters when snapshot length changes asynchronously
• In recoverBucketSnapshot(), recalculate counters after all snapshots are loaded
  to ensure accuracy (since buckets are created with snapshotLength=0 initially)
• genTopicMetricMap() reads counters without holding any lock

2. InMemoryDelayedDeliveryTracker — lock-free memory usage via delta tracking

Add AtomicLong memoryUsage that is updated by delta at each mutation point
(addMessage, getScheduledMessages, clear). getBufferMemoryUsage() returns
the cached value directly instead of iterating the nested TreeMap.

3. Dispatcher classes — remove synchronized from stats read paths

In both PersistentDispatcherMultipleConsumers and PersistentDispatcherMultipleConsumersClassic:

• getNumberOfDelayedMessages(), getDelayedTrackerMemoryUsage(),
  getBucketDelayedIndexStats(), shouldPauseDeliveryForDelayTracker() —
  removed synchronized; now use Optional.map() with the volatile field
• delayedDeliveryTracker field changed to volatile for safe publication

4. BucketDelayedMessageIndexStats — synchronized metric map generation

Add synchronized to genTopicMetricMap() to ensure atomic reads of multiple fields
when generating metrics.

5. ImmutableBucket — fix async chain in recover

Fix asyncRecoverBucketSnapshotEntry() to properly chain asyncUpdateSnapshotLength()
using thenCompose() instead of thenApply(), ensuring the snapshot length is updated
before recovery completes.

Verifying this change

• Existing unit tests in BucketDelayedDeliveryTrackerTest pass
• Manual verification: bucket count and snapshot length stats remain accurate
  across create, merge, trim, and recover operations

[lhotari]

Synchronization was explicitly added in #25681 to fix race conditions. Did you take that into account? I think that synchronization cannot be removed from shouldPauseAllDeliveries. For other methods it's most likely fine.

[lhotari]

Under a large delayed-delivery workload (~500M delayed messages)

Is this in a single topic and single subscription?

FYI, there's a limit of tracking up to 30M backlogs (with default BK nettyMaxFrameSizeBytes) when using managedLedgerPersistIndividualAckAsLongArray=true. More can be stored to metadata when using LZ4 compression, but there will be failures each time cursor state is attempted to be saved. Issue is #25985.

[nodece]

Is this in a single topic and single subscription?

Yes.

FYI, there's a limit of tracking up to 30M backlogs (with default BK nettyMaxFrameSizeBytes) when using managedLedgerPersistIndividualAckAsLongArray=true. More can be stored to metadata when using LZ4 compression, but there will be failures each time cursor state is attempted to be saved. Issue is #25985.

I will check this later.

[void-ptr974]

At a higher level, moving stats reads away from the dispatcher lock looks like the right direction for this PR. The remaining concern is making sure the newly unsynchronized stats paths do not read or mutate tracker internals without a clear concurrency boundary.

One possible short-term approach is to use the tracker lock for the small fixed-size stats sections and unsafe TreeRangeMap access, while keeping the dispatcher lock out of these stats paths. Longer term, it may be worth moving more mutable TreeRangeMap access and cached-counter maintenance behind ImmutableBucketIndex APIs, so callers do not need to touch live map views directly and stats paths can read cached/atomic values.

[void-ptr974]

I think we should separate the correctness fix from the optimization here.

The earlier discussion mentioned cached/atomic metrics as a possible direction, but the current PR is now trying to solve both synchronization correctness and metrics optimization at the same time. That adds derived state and bucket lifecycle coupling, which makes the change harder to verify.

For this PR, I would prefer to first converge on the minimal correctness fix: keep the existing bucket lifecycle behavior and collect the metrics under the tracker lock where needed. If that later proves too expensive, we can optimize the metrics path in a separate follow-up PR with dedicated invariants and tests.

[void-ptr974]

I verified this with a small Guava TreeRangeMap example.

RangeMap<Long, String> m = TreeRangeMap.create();
m.put(Range.closed(10L, 20L), "bucket");

Map<Range<Long>, String> sub =
        m.subRangeMap(Range.lessThan(15L)).asMapOfRanges();

Range<Long> clipped = sub.keySet().iterator().next();

System.out.println("original=" + m.asMapOfRanges());
System.out.println("sub=" + sub);
System.out.println("top-get-clipped=" + m.asMapOfRanges().get(clipped));

m.asMapOfRanges().remove(clipped);
System.out.println("after-exact-remove-clipped=" + m.asMapOfRanges());

Output:

original={[10..20]=bucket}
sub={[10..15)=bucket}
top-get-clipped=null
after-exact-remove-clipped={[10..20]=bucket}

So subRangeMap(...).asMapOfRanges() can return a clipped range key, while immutableBuckets.asMapOfRanges().get/remove(range) requires an exact key.

In this code path, removeBucket(range) can be a no-op, but deleteBucketSnapshot() has already deleted the snapshot and still decrements numberDelayedMessages. The tracker may then keep an immutableBuckets entry whose snapshot no longer exists and whose messages were already subtracted. Later load/trim/merge can operate on that stale bucket state.

[nodece]

Good catch! You're absolutely right about the clipped key issue.

I've fixed it in commit 4e4c709 by modifying putAndCleanOverlapRange to reconstruct the original key from the bucket object:

Range<Long> originalKey = Range.closed(bucket.startLedgerId, bucket.endLedgerId);
toBeDeletedBucketMap.put(originalKey, bucket);

This ensures removeBucket() receives exact keys instead of clipped keys from subRangeMap(), so the exact key matching works correctly.

All tests pass now, including the scenarios you identified.

[nodece]

Added regression test in e4219a0:

testOverlappingBucketsCleanupDuringRecovery verifies:

• Buckets are correctly cleaned up during recovery with overlapping ranges
• No orphaned buckets remain due to clipped key issues
• Snapshot length tracking remains consistent
• The fix prevents the specific failure mode where removeBucket() couldn't find buckets due to clipped keys from subRangeMap()

The test creates multiple buckets, closes the tracker, then recovers to ensure the bucket cleanup path works correctly.

[nodece]

Found and fixed another truncated key issue in putAndCleanOverlapRange:

Problem: The encloses() check was using truncated keys from subRangeMap(), which could incorrectly return true when the new range doesn't actually enclose the full original bucket.

Example:

• Existing bucket: [10, 30]
• New range: [15, 20]
• subRangeMap([15, 20]) returns truncated key [15, 20]
• [15, 20].encloses([15, 20]) = true ✗ (incorrect)
• Should check: [15, 20].encloses([10, 30]) = false ✓ (correct)

Fix (commits 92290bd + 37adfc0):

• Reconstruct original key from bucket object before encloses() check
• Added test testPutAndCleanOverlapRangeWithTruncatedKeys to prevent regression

This ensures buckets are only replaced when the new range truly encloses the original bucket's full range.

[void-ptr974]

Thanks for the patience through all the review iterations. This area has quite a few subtle delayed-delivery bucket edge cases, and I appreciate the follow-up fixes and regression tests.

I re-checked the latest version and don’t see a delayed-delivery correctness issue that should block this PR.

No further blocking comments from my side.

[dao-jun]

Approving - the lock-free stats reads only touch AtomicLong / primitives and never the non-thread-safe maps/bitmaps, so no blocking concerns.

One metrics-only regression worth a follow-up (or a one-liner here, since it's in code this PR already touches): totalSnapshotLengthBytes permanently inflates when a bucket is removed before its async asyncUpdateSnapshotLength completes - the thenAccept(updateBucketSnapshotLength) at line 359 is fire-and-forget, and
updateBucketSnapshotLength (line 949) has no map-membership guard, so a late length fetch adds newLength to a bucket already removed via clear/merge/trim. Pre-PR this couldn't happen since genTopicMetricMap summed the live map. Minimal fix: guard the counter update with
immutableBuckets.asMapOfRanges().containsValue(bucket) (both paths already hold synchronized(this)).

Also worth adding a test asserting bucketsCount / totalSnapshotLengthBytes match the live map across create+merge+trim+recover+clear - the current >= 0 assertion is how this slipped through.

Fixed