[fix][broker] Wait for in-flight snapshot creation when trimming orphaned delayed-delivery buckets

[lhotari]

Motivation

BucketDelayedDeliveryTracker trims orphaned bucket snapshots — buckets whose ledger range lies entirely before the cursor's mark-delete ledger — whenever sealing a new bucket pushes the bucket count past delayedDeliveryMaxNumBuckets. The trim deletes those snapshots without waiting for their snapshot creation to finish.

A bucket id is only assigned once its snapshot creation completes. Until then, ImmutableBucket.getAndUpdateBucketId() falls back to a cursor property that has not been written yet:

String bucketIdStr = ctx.cursor().getCursorProperties().get(bucketKey());
long bucketId = Long.parseLong(bucketIdStr);   // bucketIdStr == null

So deleting a bucket whose creation is still in flight throws NumberFormatException synchronously, from inside the sequential CompletableFuture chain that asyncTrimImmutableBuckets() builds. That chain deliberately stops at the first failure, so the remaining orphaned buckets are neither deleted from the snapshot storage nor removed from immutableBuckets: their snapshots leak, and the buckets stay in the range map until some later trim happens to succeed.

Every other snapshot-delete path already handles this — ImmutableBucket.clear() awaits the create future, asyncMergeBucketSnapshot() awaits the create futures of all merged buckets, and the segment-load path skips a bucket while getSnapshotCreateFuture() is not done. The trim path introduced in #25984 was the only one missing it.

This surfaced as a flaky BucketDelayedDeliveryTrackerTest.testTrimRemovesOrphanedBuckets:

java.lang.AssertionError: Remaining bucket range [1..5] should be >= 31 expected [true] but found [false]

with the cause visible in the same job's test report:

WARN  BucketDelayedDeliveryTracker - Failed to trim or merge bucket snapshots
java.util.concurrent.CompletionException: java.lang.NumberFormatException: Cannot parse null string
	at org.apache.pulsar.broker.delayed.bucket.ImmutableBucket.getAndUpdateBucketId(ImmutableBucket.java:125)
	at org.apache.pulsar.broker.delayed.bucket.ImmutableBucket.asyncDeleteBucketSnapshot(ImmutableBucket.java:322)
	at org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.deleteBucketSnapshot(BucketDelayedDeliveryTracker.java:905)
	at org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.lambda$asyncTrimImmutableBuckets$1(BucketDelayedDeliveryTracker.java:898)

Two orphaned buckets had been deleted before the chain aborted, which is why the bucket count was already within maxNumBuckets while [1..5] was still present.

The test additionally had a synchronization gap of its own: it only awaited the merging flag, which is satisfied immediately, and then asserted on immutableBuckets while the asynchronous trim was still mutating it from a snapshot-storage thread.

Modifications

• BucketDelayedDeliveryTracker.deleteBucketSnapshot() now waits for the bucket's snapshot creation to settle before deleting it, and skips the delete when the creation failed (INVALID_BUCKET_ID), since afterCreateImmutableBucket() has already removed that bucket and downgraded it to memory mode. The existing delete-and-clean-up body moved unchanged into a new private doDeleteBucketSnapshot().
• testTrimRemovesOrphanedBuckets now performs its assertions inside the Awaitility block and reads the range map under the tracker monitor, matching the existing testMergeSupportsSmallMaxNumBuckets. No assertion was weakened.
• Added testTrimWaitsForInFlightSnapshotCreation, which holds all snapshot creations in flight while the trim is triggered and then requires every orphaned bucket to be removed.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• BucketDelayedDeliveryTrackerTest.testTrimWaitsForInFlightSnapshotCreation fails deterministically without the production change (Orphaned bucket [1..5] should have been trimmed) and passes with it.
• BucketDelayedDeliveryTrackerTest is 46/46 green, including 25× invocationCount runs of both trim tests, and the NumberFormatException no longer appears anywhere in the class's output — it was also being logged by testTrimHandlesDeleteFailure.
• The whole org.apache.pulsar.broker.delayed.* package (92 tests) and ./gradlew quickCheck are green.

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