[fix][broker] Clear delayed delivery state before resetting the cursor

[nodece]

Motivation

PersistentSubscription#resetCursorInternal reset the cursor without touching the delayed delivery tracker. The tracker state is a derived cache of the backlog (the source of truth is deliverAtTime in the message metadata), and a reset moves the consumption baseline, so the whole derivation is invalidated:

• bucket snapshots, index bits and queued entries from before the reset survived into the replay, and re-added messages deduped against the stale index instead of being re-tracked cleanly;
• in-flight trims/loads/deletes from before the reset could race the replayed state;
• the bucket cursor properties stayed on the cursor, so a later tracker recovery would load pre-reset buckets.

clearBacklog and unsubscribe already clear the delayed state for the same reason; the reset path was the odd one out.

Modifications

• resetCursorInternal now clears the delayed messages and waits for the clear to settle after disconnecting consumers and before asyncResetCursor; a clear failure unfences the subscription and fails the reset.
• Without a dispatcher there is no in-flight work and the recovered bucket snapshots stay valid for the replay (re-added messages dedup through the index bitmap), so there is nothing to clear.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• BucketDelayedDeliveryTest#testResetCursorClearsDelayedMessages: tracks 100 delayed messages, resets the cursor to the earliest position and verifies the tracker count is 0 and the bucket cursor properties are removed. Without the fix the assertion fails: the tracker still reports 100 delayed messages and the bucket cursor property survives the reset.

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

Check the boxes that apply

• 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

[lhotari]

Net assessment: the change is directionally right. Clearing the delayed-delivery state before asyncResetCursor closes a real gap, the flattened chain reads much better than the old callback pyramid, and it incidentally fixes a latent hang — in the base revision a synchronous throw inside the whenComplete body (e.g. from the getTopicCompactionService() block) would have left the returned future uncompleted forever, hanging the admin request.

Two things I would like resolved before merge:

1. The dispatcher == null branch cleans nothing, so the third motivation bullet — pre-reset bucket cursor properties surviving into a later tracker recovery — is not fixed exactly when there is no dispatcher. The same method already routes the equivalent "dispatcher present but no tracker" state through cleanResidualSnapshots(cursor), and the unsubscribe path cited as precedent handles dispatcher == null explicitly.
2. clear() waits for in-flight trims but not for an in-flight bucket segment load, so the second motivation bullet ("in-flight trims/loads/deletes from before the reset could race the replayed state") is only partly achieved.

Plus one test-robustness note: the new test races the client's automatic reconnect, and one note on a silently changed failure branch.

[lhotari]

[BUG] the dispatcher == null branch leaves pre-reset bucket snapshots on the cursor, so the PR's third motivation bullet is not fixed for that case

This branch returns without cleaning anything, so a reset on a subscription with no dispatcher leaves the pre-reset #pulsar.internal.delayed.bucket* properties on the cursor — the third motivation bullet ("the bucket cursor properties stayed on the cursor, so a later tracker recovery would load pre-reset buckets") is not fixed for this case.

The asymmetry is inside this very method. When a dispatcher does exist but has no tracker yet, clearDelayedMessages() falls back to cleanResidualSnapshots(cursor) (PersistentDispatcherMultipleConsumers.java:1387-1402), which deletes the snapshots and removes the cursor properties (BucketDelayedDeliveryTrackerFactory.java:115-132). With no dispatcher at all — the same "no tracker" state — nothing runs. The unsubscribe path the description cites as precedent handles this explicitly (PersistentTopic.java:1394-1410).

Reachability. PersistentSubscription.dispatcher is never assigned null after construction (it is only created/reused in addConsumer), so dispatcher == null means "no consumer has connected since the topic was loaded" — the state right after a broker restart or a topic unload. internalResetCursorOnPosition has no guard requiring a connected consumer (PersistentTopicsBase.java:2750-2800), and resetting a cursor with consumers stopped is a common operational sequence.

Failure scenario. Bucket delayed delivery enabled; a Shared subscription has tracked delayed messages and persisted bucket snapshots; the topic is unloaded; with no consumer connected an operator resets the cursor forward. The snapshots survive the reset, and the first consumer to connect rebuilds the tracker from them (BucketDelayedDeliveryTracker.java:181-199) without filtering against the new mark-delete position — recovering already-skipped entries, inflating the delayed-message count, and retaining bucket snapshot ledgers that nothing will delete.

The description asserts this branch is safe ("the recovered bucket snapshots stay valid for the replay ... re-added messages dedup through the index bitmap"). That argument holds for a backward reset, but not for a forward one. Either route dispatcher == null through cleanResidualSnapshots(cursor) the way unsubscribe does, or narrow the claim in the description to backward resets and say why forward resets are acceptable.

[lhotari]

[INTENT MISMATCH] clear() awaits in-flight trims but not an in-flight bucket segment load, so the "in-flight loads could race" motivation is only partly addressed

The second motivation bullet says the fix stops "in-flight trims/loads/deletes from before the reset" racing the replayed state. Awaiting clearDelayedMessages() here covers trims and deletes, but not loads.

BucketDelayedDeliveryTracker.clear() (:795-816) chains only off trimFuture. The in-flight segment load tracked in pendingLoad (declared at :131, assigned at :717) is never awaited, and its continuation (:718-734) re-populates snapshotSegmentLastIndexMap and sharedBucketPriorityQueue. Both that continuation and the body of clear() (:806) take the tracker monitor, so they are mutually exclusive but unordered.

Failure scenario. A bucket segment load is already in flight when the reset disconnects consumers. clear() empties the queues and cleanImmutableBuckets() removes the immutable buckets; the outstanding load's continuation then re-inserts pre-reset indexes into sharedBucketPriorityQueue and re-registers a bucket that was just removed. asyncResetCursor proceeds against a tracker that is not actually empty, which is the state this PR set out to prevent.

To be fair on scope: this is a pre-existing property of clear() (shared with clearBacklog and unsubscribe), not a regression introduced here, and the window is narrow — consumers are already disconnected by this point, so only a load started before the reset can land. But since the stated purpose of the change is to make the reset wait for the delayed state to settle, it is worth either awaiting pendingLoad in clear() or narrowing the second motivation bullet.

[lhotari]

[QUALITY] the new test races the client's automatic reconnect, which can re-track the delayed messages before the assertions run

These two assertions are unguarded, and the comment above the reset states the no-re-tracking assumption without enforcing it.

The reset disconnects the consumer via Consumer.disconnect (Consumer.java:496-504), which sends CommandCloseConsumer. The client reconnects after initialBackoffIntervalNanos, default 100 ms (ClientConfigurationData.java:331-334, ClientCnx.java:1284-1300). PersistentSubscription.addConsumer (:248-251) explicitly chains the reconnect behind inProgressResetCursorFuture, so the reconnect lands immediately after the reset completes — exactly where these assertions run. Once resubscribed, the dispatcher reads from the reset position and re-tracks the 100 delayed messages, restoring both the tracker count and the bucket cursor properties.

Failure scenario. On a loaded CI runner, the reset itself (bucket snapshot deletes over metadata plus the cursor ledger write) plus scheduling jitter exceeds the ~100 ms backoff that started when the consumer was disconnected at the top of the reset. The consumer resubscribes and re-tracks before this line runs, so assertEquals(dispatcher.getNumberOfDelayedMessages(), 0) sees 100, or bucketKeysAfterReset is non-empty.

Suggested fix: consumer.close() before admin.topics().resetCursor(...). PersistentSubscription.dispatcher is never nulled, so dispatcher != null still holds and clearDelayedMessages() is still exercised; dispatcher.isConsumerConnected() is then false, so disconnectFuture completes immediately and the rest of the path is unchanged.

Worth adding while you are here: the test covers neither the dispatcher == null branch nor the clear-failure branch that the description says now unfences the subscription.

[lhotari]

[QUALITY] the whenComplete -> thenCompose rewrite silently dropped the disconnect-failure branch (resetCloseFuture skipped, 412 becomes 500) — currently unreachable, flagging so the deletion is deliberate

Moving from whenComplete to thenCompose changed what happens when disconnectFuture completes exceptionally. The base revision ran, on either outcome:

if (dispatcher != null) { dispatcher.resetCloseFuture(); }   // unconditional
if (throwable != null) {
    ...
    future.completeExceptionally(
        new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
    return;
}

thenCompose skips its body when the upstream future fails, so resetCloseFuture() no longer runs — leaving a stale closeFuture on the dispatcher (PersistentDispatcherMultipleConsumers.java:657-659) — and the terminal .exceptionally surfaces BrokerServiceException instead. That would change the admin response: SubscriptionBusyException maps to 412 PRECONDITION_FAILED (PersistentTopicsBase.java:2816-2822 and :2524-2530), while anything else falls through resumeAsyncResponseExceptionally to 500.

This is currently unreachable, so it is a latent semantics change rather than a live bug: in every persistent dispatcher closeFuture is only ever completed with complete(null) (PersistentDispatcherMultipleConsumers.java:631-641, PersistentDispatcherMultipleConsumersClassic.java:541-553, AbstractDispatcherSingleActiveConsumer.java:337-343). I am flagging it only so the removal is deliberate rather than incidental to the reflow — if the branch is genuinely dead, dropping it is fine, but it would be good to say so in the PR description.