[fix][broker] Prevent Key_Shared out-of-order replay starvation at the end of topic

[Denovo1998]

Motivation

Key_Shared out-of-order replay could become starved at the end of a topic when the replay queue began with messages assigned to consumers without available permits. This prevented later eligible messages from being dispatched.

Modifications

• Preserve position-to-sticky-key hash mappings for replayed messages in both delivery modes.
• Apply consumer permit checks during out-of-order replay.
• Restrict look-ahead triggering to cases where the cursor has more entries.
• Add regression coverage for ordered and out-of-order delivery modes.

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

[lhotari]

Review summary

Reviewed against 524cc44 with full repository context. No correctness or security problem found — everything below is quality or test coverage. The items I'd most like to see addressed before merge are the unconditional map allocation, the test not discriminating the change it covers, and the now-stale comments in the classic dispatcher.

On the central question: is it safe to remove the #26236 loop brake?

Yes, as far as I can trace. #26236 kept out-of-order look-ahead unconditional deliberately, as a brake against repeated read-and-discard cycles, because the out-of-order replay queue tracked no sticky-key hashes and the replay filter therefore could not exclude messages for consumers without permits. This PR removes that brake and compensates by recording position -> hash in out-of-order mode too. The compensation looks complete:

1. Every dispatch-time discard re-enters the replay queue with a real hash. filterAndGroupEntriesForDispatching re-adds via addMessageToReplay(ledgerId, entryId, stickyKeyHash), and that hash can never be the sentinel: StickyKeyConsumerSelectorUtils.makeStickyKeyHash remaps 0 to 1, and the PIP-486 entry-bucket path nudges 0 to 1 as well. So a position can take the filter's "hash unknown" branch at most once.
2. Map is a subset of the bitmap. The three-arg add writes both; the two-arg add writes only the bitmap and never clears the map; remove, removeAllUpTo and clear clear both consistently. The "unknown hash" state cannot be re-entered.
3. Zero-progress rounds are backoff-paced, not spun. At the end of the topic a fully discarded read leaves lastNumberOfEntriesProcessed == 0 and skipNextBackoff == false (its only setter is inside the hasMoreEntries() branch), so handleSendingMessagesAndReadingMore takes reScheduleReadWithBackoff() with an increasing backoff rather than re-entering readMoreEntries() on the same thread.
4. The terminal state is the one the brake used to force, minus the starvation. Once the undispatchable positions carry hashes, getMessagesToReplayNow returns empty and readMoreEntries falls through to a normal read that waits at the end of the topic (or pauses at the permits gate). A parked normal read does not block replay — replay selection happens before the doesntHavePendingRead() guard — so consumerFlow recovery dispatches immediately.

The two hash-less entry points, delayed-tracker due messages and redeliverUnacknowledgedMessages(consumer, positions) (the one with the standing TODO about the missing hash), each cost exactly one extra replay read before converging — the same thing the ordered path has always done.

I also confirmed the starvation is real on the base commit: with the old unconditional out-of-order filter, getMessagesToReplayNow returns the first N bitmap positions, and if those all belong to a permit-less consumer the whole batch is discarded and the brake parks a normal read, leaving later eligible replay messages stuck until an unrelated event. In-order and plain-Shared dispatch paths are unchanged by this diff.

Point 1 is load-bearing and implicit. If a real sticky-key hash could ever be 0, the new out-of-order sentinel early-return in add() would reopen exactly the cycle the brake existed to stop. A short comment recording that invariant would be worth having.

Two findings that don't map onto a changed line

Stale comments in PersistentStickyKeyDispatcherMultipleConsumersClassic. Lines 557 and 606 both read // The variable "hashesToBeBlocked" and "recentlyJoinedConsumers" will be null if "isAllowOutOfOrderDelivery()". After this rename the field is positionToStickyKeyHash and it is now never null in any mode. That makes the out-of-order short-circuit at line 559 look like a null-safety guard when it is now purely a semantic skip — a later cleanup acting on the comment would silently change classic out-of-order behaviour. Worth updating both comments in this PR since it is the rename that invalidates them.

Related, and cheap to fix: classic Key_Shared out-of-order now populates positionToStickyKeyHash with real hashes, but the only classic consumer of getHash() is filterOutEntriesWillBeDiscarded, which returns early for out-of-order at lines 559-561 before ever reaching the getHash() call at line 570. So classic out-of-order pays map memory plus the removeAllUpTo scan for data nothing reads. An allowOutOfOrderDelivery && isClassicDispatcher early-return in add() would avoid it.

The description doesn't mention #26236. "Restrict look-ahead triggering to cases where the cursor has more entries" reads as a neutral tightening, but the allowOutOfOrderDelivery || term and its ten-line justification comment were added by #26236 specifically as a loop brake, and this PR deletes both. The reason the brake is no longer needed — points 1-4 above — is the crux of the change and belongs in the Modifications section, so a future bisect landing here has the reasoning.

---

_{Assisted-by: Claude (Opus 5) and Codex (gpt-5.6-sol). Findings were produced by independent reviews from both models over the full checkout, then cross-validated by each model against the other's conclusions; every finding below is anchored to a verified code path. Attribution per the ASF Generative Tooling guidance.}

[lhotari]

Memory: this map is now allocated for every MessageRedeliveryController, including dispatchers that can never populate it.

The plain Shared dispatcher builds the controller with allowOutOfOrderDelivery = true (PersistentDispatcherMultipleConsumers ctor) and its getStickyKeyHash always returns STICKY_KEY_HASH_NOT_SET, so every add() takes the new early-return below and this map stays empty forever. Before this PR it was null in that mode.

The cost is fixed and non-trivial: expectedItems(128) / concurrencyLevel(2) -> 64 per section -> (int)(64 / 0.66) = 96 -> alignToPowerOfTwo -> capacity 128 -> new long[ITEM_SIZE * 128] = 4 KiB per section, 8 KiB per controller. autoShrink can't reclaim it either — shrink is floored at initCapacity. So ~78 MiB at 10k Shared subscriptions, ~780 MiB at 100k, all of it never written to.

Allocating lazily (or only when the dispatcher actually records hashes) would keep the fix free for Shared and classic-out-of-order subscriptions. This also subsumes the removeAllUpTo nit below.

[lhotari]

Minor: dropping && !allowOutOfOrderDelivery means that whenever removeUpTo actually clears bits, this now allocates an ArrayList and walks both sections' full 512-long tables — including for Shared subscriptions where the map is provably empty (see the comment on the constructor). ConcurrentLongLongPairHashMap.forEach has no empty short-circuit.

To be fair on severity: for out-of-order Key_Shared this scan is now necessary — the recorded hashes have to be released — so that part is the price of the fix and parity with what ordered mode has always paid, not an avoidable regression. Only the always-empty-map case is avoidable, and it's sub-microsecond per call. A && !positionToStickyKeyHash.isEmpty() guard, or lazy allocation, covers it.

[lhotari]

This comment and the skipNextBackoff = cursor.hasMoreEntries(); on the next line are now dead logic. The branch is only entered when triggerLookAhead.booleanValue() && cursor.hasMoreEntries() (line 331), it's the same synchronized block with no cursor mutation in between, and the read position doesn't move here — so the re-check is always true.

Under the old (allowOutOfOrderDelivery || cursor.hasMoreEntries()) guard the inner re-check was meaningful for the out-of-order case; after this change it isn't. skipNextBackoff = true; is equivalent and clearer, and the comment can go.

[lhotari]

The stickyKeyHash == STICKY_KEY_HASH_NOT_SET half of this condition is unreachable for this (non-classic) dispatcher: in ordered mode MessageRedeliveryController.add throws on the sentinel, in out-of-order mode it early-returns without storing it, the two-arg add stores nothing at all, and real hashes can never be 0 (makeStickyKeyHash remaps 0 to 1; the entry-bucket path nudges 0 to 1).

No objection to keeping it as defence-in-depth, but a short "unreachable today, guards against a future selector that can emit 0" note would save the next reader the derivation. (For what it's worth, this is the same invariant the whole convergence argument rests on — see the review summary.)

[lhotari]

!allowOutOfOrderDelivery && is redundant here: alreadyBlockedHashes is only ever populated under !allowOutOfOrderDelivery (the three add sites below), so in out-of-order mode the set is provably empty and contains is already false.

Not wrong — at best it saves a hash lookup on an empty set — but combined with the two other new guards it makes the reader re-derive which branches are live. Hoisting !allowOutOfOrderDelivery once, or funnelling the three alreadyBlockedHashes.add(...) sites through a small blockHash(hash) helper, would state the intent directly: the blocked-hash machinery exists only to preserve ordering, so out-of-order skips it.

[lhotari]

This guard is dead: drainingHashesRequired is keySharedMode == AUTO_SPLIT && !allowOutOfOrderDelivery, so !allowOutOfOrderDelivery is guaranteed true whenever the enclosing block is entered (and drainingHashesTracker is null in out-of-order mode anyway).

Harmless, but it reads as if draining could interact with out-of-order delivery, which it can't.

[lhotari]

Inserting the @DataProvider here detaches the Javadoc above — the block explaining issue #21554 and the stall scenario now documents allowOutOfOrderDelivery() instead of the test it was written for.

Moving the data provider above the Javadoc (or below the test) keeps the explanation attached to testLookAheadNotEngagedWhenCursorHasNoMoreEntries.

[lhotari]

This variant doesn't discriminate the changes it's meant to cover. Of the three production changes in this PR, the new allowOutOfOrderDelivery = true run only pins down one of them:

• Look-ahead condition (line 331) — covered. Revert it and cycle 2 skips replay, parks at the no-op asyncReadEntriesWithSkipOrWait mock, and the latch times out.
• The permit check in ReplayPositionFilter — not covered. Revert it to the old unconditional return true for out-of-order and cycle 2 replays all of {1:1, 1:2, 1:3}; entries 1-2 are discarded (slow consumer at 0 permits) but entry 3 still reaches consumerMock (1000 permits, read batch 100, cap mock is identity), so the latch still fires and the test still passes.
• Recording hashes in out-of-order mode — also not covered. Revert it and getHash returns null, the filter's hash-missing branch passes everything, and the test is still green.

Nothing asserts the property that actually replaces the deleted loop brake: that entries 1-2 stop being re-read once their consumer hits zero permits. Bounding asyncReplayEntries invocations, or asserting the second replay read excludes 1:1/1:2, would pin it.

Two more gaps worth a line each:

• The hash-less path the filter explicitly exempts is untested in out-of-order mode — everything here is seeded through addEntryToReplay, which always records a hash, and delayed delivery is disabled in the mock. Production feeds due delayed messages (and redeliverUnacknowledgedMessages(consumer, positions)) through the two-arg add, with no hash.
• MessageRedeliveryControllerTest never exercises the new out-of-order + STICKY_KEY_HASH_NOT_SET early-return branch of add(long, long, long).

To reproduce the starvation the Motivation actually describes, the read budget has to be the scarce resource — constrain maxMessagesToRead so the undispatchable positions would consume it, then assert the eligible position is still selected.