[fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures

[void-ptr974]

Motivation

BucketDelayedDeliveryTracker publishes a newly sealed immutable bucket before its asynchronous snapshot creation completes. If the same addMessage call causes the bucket count to exceed delayedDeliveryMaxNumBuckets, trim can immediately select a bucket whose snapshot ID is not available yet. This makes the current trim pass fail and delays cleanup until a later trim is triggered.

Modifications

Consider a tracker with a maximum of 5 buckets and a first active ledger ID of 50:

1. Rapid message additions create 6 immutable buckets while their snapshot creations are still in flight. The same addMessage call that creates the sixth bucket starts trim because the limit has been exceeded.
2. Since the bucket ranges end before ledger 50, trim considers them orphaned and attempts to delete them sequentially.
3. Before this change, deletion starts immediately. For an in-flight snapshot, the bucket ID is neither cached in the bucket nor recorded in the cursor properties, so getAndUpdateBucketId() calls Long.parseLong(null) and throws NumberFormatException. The trim chain stops at that failure and trimFuture completes exceptionally. It is not permanently pending, but the failed deletion is left for a later trim attempt.
4. With this change, each deletion is chained from the bucket's snapshot creation future. After creation completes, trim revalidates under the tracker lock that the bucket ID is valid, the range is still before the current first active ledger, and the exact range still maps to the same bucket instance.
5. Only a bucket that still passes those checks is deleted. After deletion succeeds, the mapping is checked again before removing the bucket indexes and updating the delayed-message count, preventing an older asynchronous continuation from modifying newer tracker state.

The wait is asynchronous and does not block the dispatcher thread.

Verifying this change

Added deterministic tests that hold snapshot creation in flight and verify that:

• Trim does not delete snapshots before creation completes and resumes after creation is released.
• Trim revalidates bucket eligibility after waiting and does not delete based on stale state.

Verified with:

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTrackerTest
• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTrackerThreadSafetyTest
• ./gradlew quickCheck

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

• 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]

The orphan eligibility is validated only before the asynchronous snapshot deletion begins. Once this synchronized block is released, the cursor may be reset or its mark‑deleted position could move backward while asyncDeleteBucketSnapshot() is still running.

In such cases, the completion block only rechecks the range‑to‑bucket identity, so it may still delete bucket state even if range.upperEndpoint() < firstActiveLedgerId() no longer holds. The existing test updates firstLedgerId before the snapshot‑create future is released, but it does not cover modifications after this validation and after storage deletion has started.

Could we add a deletion ownership or version guard that spans the duration of the asynchronous operation, or base orphan cleanup on a monotonic physical‑ledger boundary? Please also include a test that blocks deleteBucketSnapshot(), changes firstLedgerId after deletion begins, and then completes the delete.

[Denovo1998]

This ensures the global trimFuture waits for every selected bucket's snapshot-creation future. If any create future remains pending for an extended period, trimFuture.isDone() stays false, which prevents later addMessage() calls from triggering another trim/merge and also chains clear() behind the same future.

Would it be safer to skip buckets still in the CREATING state and retrigger trim when their creation finishes, rather than holding the global trim/merge gate? At minimum, it would be helpful to document the bounded-completion guarantee or add a test for a create future that never completes.

[nodece]

@void-ptr974 #26280 should include this fix. When fixing the trim and segement loding, I found this issue.