[improve][broker] Trim orphaned bucket snapshots when ledgers are deleted

[dao-jun]

Motivation

When a topic's managed ledger trims old ledgers (via retention or compaction), the immutable bucket snapshots in BucketDelayedDeliveryTracker that cover those deleted ledger ranges become orphaned — their underlying message data no longer exists, yet they:

1. Continue to occupy storage as bucket snapshots.
2. Count against the maxNumBuckets limit, potentially triggering unnecessary and expensive merge operations.

Prior to this change, the tracker only merged buckets when the count exceeded the limit; it never cleaned up buckets whose ledger range had been fully trimmed. This change introduces a trim step that runs before merge, deleting orphaned buckets and decrementing the delayed-message counter accordingly.

Modifications

Source (BucketDelayedDeliveryTracker.java)

• Added trimFuture: A volatile CompletableFuture that represents the currently in-flight trim/merge/clear operation. It acts as both a gate (preventing concurrent trim chains) and a synchronization point for clear().
• Added asyncTrimImmutableBuckets(): Scans the managed ledger's surviving ledger range. Buckets whose upperEndpoint falls entirely before the earliest surviving ledger are selected for deletion — these are orphaned and safe to remove.
• Added deleteBucketSnapshot(): Removes a single bucket from immutableBuckets, asynchronously deletes its snapshot, and re-adds the bucket on failure (rolling back the message count). Failures propagate via CompletionException to stop the sequential deletion chain, avoiding wasted attempts when storage is unavailable.
• Changed the merge trigger in addMessage(): Before merging, the tracker now runs asyncTrimImmutableBuckets() first. The trimFuture gate ensures only one trim/merge chain is in-flight at a time.
• Added early-return to asyncMergeBucketSnapshot(): If the bucket count is already within the limit, merge returns immediately without scanning for merge candidates.
• Rewrote clear(): Instead of immediately clearing state (which could race with an in-flight trim's failure callback), clear() now chains itself after any pending trimFuture. It also sets trimFuture to the clear chain, blocking new trim triggers until the clear completes.

Test (BucketDelayedDeliveryTrackerTest.java)

Added 4 tests using a helper that injects a mocked ManagedLedger:

Test	Scenario
testTrimRemovesOrphanedBuckets	Buckets covering ledgers below firstLedgerId are removed, remaining ones have ranges ≥ firstLedgerId
testTrimHandlesDeleteFailure	Injected delete exception is consumed, tracker remains operational
testTrimWithNoOrphanedBuckets	With firstLedgerId=0, trim is a no-op, merge runs normally
testMergeEarlyReturnWhenWithinLimit	With maxNumBuckets far above actual count, merge returns early without compaction

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

[void-ptr974]

I think there is still an issue here: getScheduledMessages() only skips entries before firstLiveLedgerId when !containsMessage(...).

For old-ledger entries that are still present in the bucket bitmap, containsMessage(...) returns true, so those deleted-ledger positions can still be returned. The key point is that ledgerId < firstLiveLedgerId should be enough to skip the entry, and the tracker state/counter should stay consistent.

addressed

[lhotari]

#26080 is also required when cherry-picking