[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

[Denovo1998]

Removing a bucket from immutableBuckets might not be enough. When a mutable bucket is sealed, the first snapshot segment has already been loaded into sharedBucketPriorityQueue, and snapshotSegmentLastIndexMap can still point to this bucket. After this trim removes the range and decrements numberDelayedMessages, getScheduledMessages() can still pop those queued entries and return positions from ledgers that have already been deleted, then decrement numberDelayedMessages again.

Could we either purge the in-memory state for the trimmed bucket under the same lock, such as rebuilding sharedBucketPriorityQueue and removing matching snapshotSegmentLastIndexMap entries, or otherwise avoid trimming buckets that still have loaded segments? Please also add a test that advances the clock after trim and asserts no positions below firstLedgerId are returned and the delayed-message counter remains consistent.

[dao-jun]

Not clearing the data of sharedBucketPriorityQueue is intentionally designed, as ManagedCursorImpl.asyncReplayEntries will filter out invalid Positions.

I did miss cleaning the snapshotSegmentLastIndexMap, this is a problem.

Double decrease numberDelayedMessages is indeed a problem, good catch.

[Denovo1998]

If clear() is called while trimFuture is still in flight, and that trim/merge chain later completes exceptionally, this before.thenCompose(...) will not execute the clear block. The whenComplete in addMessage only logs the failure and preserves the exceptional completion, so clear() can return exceptionally while leaving immutableBuckets, sharedBucketPriorityQueue, lastMutableBucket, snapshotSegmentLastIndexMap, and numberDelayedMessages uncleared.

Could we normalize the previous future before chaining clear, for example with handle/exceptionally, so clear always runs after the in-flight trim/merge settles?

[dao-jun]

Good catch! That’s a bug for sure.

[Denovo1998]

This test only checks that immutableBuckets no longer contains ranges below firstLedgerId. It does not verify the externally visible behavior after trim. Please advance the mock clock and call getScheduledMessages(), then assert that no position from ledgers below firstLedgerId is returned and that getNumberOfDelayedMessages() remains consistent. That would catch stale entries left in sharedBucketPriorityQueue after the bucket range is removed.