[fix][broker] Manage in-flight bucket work during clear

[nodece]

Motivation

The bucket delayed delivery tracker can complete clear() while bucket work is still in flight:

• clear() reset in-memory state without waiting for in-flight snapshot segment loads, so a completing load callback could republish pre-clear entries.
• Snapshot deletes were partly fire-and-forget: merge source-bucket deletes were not folded into the merge future, and recovery/terminal-load deletes were not tracked, so clear() could return while those deletes were still running.

This is the in-flight-future subset of #26280. Data-ownership races around trim eligibility, same-range bucket replacement, stale cursor-property removal, and clear fencing require lifecycle/revalidation changes and will be handled in a follow-up PR.

Modifications

• clear() now waits for the in-flight trim/merge chain (trimFuture), pending snapshot segment loads (pendingLoad), and tracked snapshot deletes (pendingDeletes) before resetting in-memory state.
• Terminal-segment-load deletes are chained into pendingLoad; recovery and trim deletes are tracked through pendingDeletes.
• The merge path folds source-bucket snapshot deletes into its returned future so trimFuture covers them.
• Added deterministic tests for segment load, terminal-load delete, merge create/load/delete, tracked-delete failure, and late length updates.

[nodece]

Ping @void-ptr974 @Denovo1998

[void-ptr974]

Thanks for the fix. I think there are two edge cases that may still need attention.

1. Merge snapshot creation failure followed by clear/restart

For example, suppose buckets [1,5] and [6,10] are being merged into [1,10]. If creating the [1,10] snapshot fails, the source bucket delete chain is skipped. At that point [1,5] and [6,10] may already have been removed from immutableBuckets, but their cursor properties can still remain.

A later clear() only cleans the buckets currently reachable from immutableBuckets, so those source bucket properties may not be removed. If the tracker is recreated with the same cursor/storage, the old [1,5] and [6,10] buckets can be recovered again.

2. Conditional cursor-property removal with BadVersion retry

For example, suppose a stale delete is deleting bucket key #pulsar.internal.delayed.bucket_1_5 with old bucket id 0. removeBucketCursorPropertyIfCurrent() first reads the property and sees 0, so it proceeds to remove it. Before the remove succeeds, another operation recreates the same bucket key with new bucket id 999. The first remove attempt then hits BadVersionException.

The retry is only for removeCursorProperty(), so it does not re-read the current value. It can remove the property even though it now points to bucket id 999, which means the recreated bucket cursor property can still be removed.

[Denovo1998]

The resolveBucketIdForDelete() method may wait for the snapshotCreateFuture, but the orphan decision is already captured earlier in asyncTrimImmutableBuckets(). This continuation does not revalidate either condition—whether range.upperEndpoint() < firstActiveLedgerId() or immutableBuckets.asMapOfRanges().get(range) == bucket—before deleting or removing tracker state.

As a result, a cursor reset, a backward mark-delete movement, or a same-range bucket replacement while bucket creation is pending could delete or untrack live delayed indexes. The removeBucket(range) operation may also remove a replacement bucket because it does not verify object identity. This race condition is described in #26260/#26280. However, this PR is based independently on master and does not include the reset fencing introduced in #26280.

Could we revalidate eligibility and the exact range-to-bucket identity before initiating storage deletion, and again before mutating tracker state at lines 942-945?

synchronized (this) {
    snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
    removeBucket(range);
    bucket.getDelayedIndexBitMap().forEach((ledgerId, bitmap) ->
            bitmap.forEachLong(entryId -> index.untrack(ledgerId, entryId)));
}

For destructive trim operations, using the managed ledger’s monotonic retained-ledger boundary would also prevent mark-delete from moving backward. Please add a deterministic test that simulates a blocked create operation followed by a cursor reset or bucket replacement.

[nodece]

Thanks for the detailed review. This PR is now scoped to in-flight future management only. I agree that trim eligibility and exact range-to-bucket identity need revalidation before destructive storage deletion and before mutating tracker state; I am splitting that ownership/revalidation fix into a follow-up PR together with the reset fencing discussion from #26280. The current change keeps the trim delete inside the merge/trim future so clear() waits for it, and the follow-up will add deterministic tests for blocked creation followed by reset or same-range replacement.

[Denovo1998]

The expected-bucket-ID check is located outside the executeWithRetry supplier. Since ManagedCursorImpl persists cursor properties using optimistic versioning, a concurrent writer from another tracker or sequencer may update this property after the check, causing the first removal attempt to fail with a BadVersionException.

Then, in a retry, executeWithRetry only calls removeCursorProperty(bucketKey) again, without revalidating the expected ID. This retry could inadvertently delete a replacement bucket’s property. Sharing the same BucketContext sequencer does not safeguard against concurrent writes from a new tracker or direct cursor‑property modifications.

Should we place the expected‑value check inside the retry supplier, or introduce an atomic compare‑and‑remove cursor‑property operation? The regression test should update the property so the first removal returns BadVersionException, then confirm that the retry avoids deletion.

[nodece]

Agreed — the expected-value check must be retried together with removeCursorProperty(), otherwise a BadVersion retry can delete a replacement property. The unconditional/stale cursor-property cleanup and atomic compare-and-remove behavior belong to the data-ownership follow-up, so I have removed that item from this PR's scope. The current PR only ensures deletes are represented in the future chain waited on by clear()/trim.

[Denovo1998]

This takes a one-time snapshot of pendingLoad and pendingDeletes, but clear() does not mark the tracker as clearing or prevent getScheduledMessages() from starting another load.

While clear() waits for trimFuture or a captured load, the dispatcher or timer may still enter the tracker. Once the captured load completes, another call could acquire the tracker lock, see that pendingLoad is finished, and schedule the next load before the cleanup routine runs. That new callback could then repopulate queued and mapped items after lines 835–838 clear them. shouldPauseAllDeliveries() would also remain false.

Could clear() establish a lifecycle generation or clearing state synchronously on entry, prevent new loads during clearing, and make every async callback verify the generation before publishing its state?

[nodece]

Thanks. This PR intentionally fixes the quiescence snapshot: clear() now waits for trimFuture, the current pendingLoad, and tracked pendingDeletes before resetting state, with tests for segment load, terminal-load delete, merge create/load/delete, and failed tracked deletes. A full clearing lifecycle/generation fence is broader than inflight-future management and is planned for the follow-up PR; it will prevent newly scheduled loads from starting during clear and verify generation before publishing callbacks.

[Denovo1998]

I re‑checked the production call path. PersistentSubscription.clearBacklog() calls dispatcher.clearDelayedMessages() while the dispatcher remains usable, and the normal dispatcher path can still call getScheduledMessages().

Thus, a single snapshot of pendingLoad / pendingDeletes does not serve as a quiescence barrier: after the captured load completes, another load can begin before the cleanup continuation acquires the tracker lock, and that callback could publish state after clear() has finished.

This appears to fall directly within the guarantee stated in this PR, not just as a broader reset/ownership issue. Could clear() establish a clearing generation or state before releasing the lock, prevent new async work from starting, and require pre‑clear callbacks to verify that generation before publishing?

[nodece]

I'd still keep this one out of this PR. What this change promises is that clear() drains everything that was in flight when it entered. A load scheduled after that is new-lifecycle work, and what it publishes after the reset is inert — entries get popped with removeIndexBit() returning false and silently dropped (never delivered), and the boundary entries point at buckets that are no longer in immutableBuckets, so they can't chain more loads. It burns a few wasted reads and stops by itself. Master today doesn't drain anything at all, so this is strictly narrower already.

Doing what you suggest properly — generation + blocking new work, verified at publish — is a real lifecycle change of its own, and I'd rather not bolt it onto this one. I'll take it in the follow-up, happy to discuss the shape there.

[Denovo1998]

The source-deletion chain is gated by the post-processed snapshotCreateFuture; however, afterCreateImmutableBucket() converts a failed snapshot creation into a normally completed INVALID_BUCKET_ID. Consequently, this thenCompose still deletes all source snapshots even when the merged snapshot was never created. Because the source bucket mappings were previously removed, a restart following this path can lose the only durable delayed-index state.

There is also a second variant: ImmutableBucket.asyncSaveBucketSnapshot() currently suppresses putBucketKeyId() failure and still completes with the new bucket id, so a positive id does not guarantee the merged snapshot is recoverable.

Could source deletion be gated on successful durable handoff — including both snapshot persistence and cursor-property publication — while keeping the source buckets recoverable otherwise?

[nodece]

Agreed, the delete was wrong — fixed in 9781754: the source-delete chain now skips when the merged creation completed with INVALID_BUCKET_ID, so the source buckets stay recoverable from their cursor properties. Added a test that fails the merged creation and recreates the tracker with the same cursor/storage: both source buckets come back.

The second variant — putBucketKeyId() failing but still completing the create future with a positive id — is create-path semantics from master; changing it here felt too invasive. Even in that case delivery is not at risk: deliverAtTime is in the message metadata, and undelivered messages are just re-read from the backlog and re-tracked on restart (the same fallback the in-memory tracker uses). That part fits the follow-up / #26280.

[Denovo1998]

I think clear() may still return success even when some snapshot cleanup actually failed.

asyncDeleteBucketSnapshot() only removes the cursor property after the snapshot delete succeeds. But in both the terminal-load path and merge path, the bucket is already removed from immutableBuckets before the delete finishes.

So if the delete keeps failing after retries, clear() ignores the error here and continues. At that point cleanImmutableBuckets() also can't find this bucket anymore. The old cursor property may still be there, and after recreating the tracker the old bucket could be recovered again.

The current tests only check that clear() completes successfully. Could we also recreate the tracker with the same cursor/storage and verify that the deleted bucket isn't recovered again?

I think either clear() should propagate this cleanup failure, or we need to keep track of the failed cleanup until its cursor property is actually removed.

[nodece]

Looked at both failure shapes, I think they both converge:

• delete keeps failing: snapshot and property survive as a consistent pair — recovery just loads the bucket again (it still exists) and the next trim/clear retries the cleanup.
• delete ok but property removal failed: recovery resolves the property, hits BucketNotExist, and the toBeDeleted path in recoverBucketSnapshot() removes it.

While writing the recreate test for the second shape I found that path was actually dead: the chain delivers the BucketNotExistException wrapped in a CompletionException, so the instanceof never matched and recovery would just throw. Fixed by unwrapping before the check (e3c264a), and the test now recreates the tracker with the same cursor/storage and watches the stale property get cleaned up — plus a test for the consistent-pair shape, where the bucket just loads normally again.

[Denovo1998]

I re‑checked the production call path. PersistentSubscription.clearBacklog() calls dispatcher.clearDelayedMessages() while the dispatcher remains usable, and the normal dispatcher path can still call getScheduledMessages().

Thus, a single snapshot of pendingLoad / pendingDeletes does not serve as a quiescence barrier: after the captured load completes, another load can begin before the cleanup continuation acquires the tracker lock, and that callback could publish state after clear() has finished.

This appears to fall directly within the guarantee stated in this PR, not just as a broader reset/ownership issue. Could clear() establish a clearing generation or state before releasing the lock, prevent new async work from starting, and require pre‑clear callbacks to verify that generation before publishing?