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

[Denovo1998]

Motivation

PR #26236 deliberately kept look-ahead enabled when allowOutOfOrderDelivery was true to prevent repeated read-and-discard cycles. At that time, the out-of-order replay queue did not retain position-to-sticky-key-hash mappings, so replay selection could not skip entries assigned to consumers without available permits.

At the end of a topic, that loop brake can cause the next cycle to skip replay and park a normal read while a later replay entry is still eligible for delivery. Dispatch then remains stalled until an unrelated event, such as a consumer flow request, triggers another read.

This change removes that starvation while preserving the convergence property introduced by #26236. Known replay hashes are retained and checked against consumer permits before spending the replay read budget. Hash-less positions are still admitted once and, if rejected during dispatch, are requeued with their calculated hash so subsequent replay cycles can filter them.

Modifications

• Retain position-to-sticky-key-hash mappings for modern out-of-order Key_Shared replay entries with known hashes.
• Lazily allocate the position-hash map so plain Shared, classic out-of-order, and hash-less replay paths do not pay its fixed memory cost.
• Filter known out-of-order replay positions by their selected consumer's available permits.
• Trigger look-ahead only when the cursor has more entries, preventing a normal read from being parked at the end of the topic while eligible replay entries remain.
• Preserve classic out-of-order replay behavior and simplify redundant replay-filter conditions and comments.
• Add deterministic coverage for the constrained replay read-budget scenario, hash-less and sentinel paths, lazy allocation, and sticky-key ref-count cleanup.

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]

@Denovo1998 Please check the review comments. It would be great to get this fix PR merged.

[Denovo1998]

@lhotari Sorry, I was delayed by some things recently.
I just took another look and made some changes, please review again.

[lhotari]

Re-reviewed 711339c against the full checkout. All 8 review comments and both summary findings from the first round are addressed — replied inline and resolved each thread. No new findings.

The two items I most wanted fixed are both solid:

• Lazy allocation covers more ground than I asked for. Plain Shared and classic Key_Shared out-of-order now allocate nothing, which also removes the map-plus-removeAllUpTo-scan that classic out-of-order was paying for data that filterOutEntriesWillBeDiscarded never reads. Ordered mode still allocates in the constructor, so hashesRefCount accounting is untouched by the new path.
• The tests now discriminate. I checked this by mutation rather than by reading — reverting each of the three production changes individually turns a test red, where on 524cc44 only the look-ahead condition was pinned. Details in the thread.

Also worth noting: the stickyKeyHash == STICKY_KEY_HASH_NOT_SET invariant that the whole convergence argument rests on is now written down in MessageRedeliveryController.add(), where it is established rather than where it was consumed, and the Motivation section explains why #26236's loop brake can go. That is the part a future bisect will need.

Local verification on 711339c:

• MessageRedeliveryControllerTest — 8 tests, 0 failures
• PersistentStickyKeyDispatcherMultipleConsumersTest — 23 tests, 0 failures
• ./gradlew quickCheck — BUILD SUCCESSFUL

One thing to watch: GitHub reports no check runs on 711339c yet, so CI has not exercised this revision. Worth getting a green run before merge, since the add() restructuring touches all four (ordered / out-of-order) × (classic / non-classic) modes plus plain Shared, and the broader Key_Shared suites are what would catch a mode mix-up.

LGTM once CI is green.

_{Assisted-by: Claude (Opus 5). Verification was local: full-checkout re-read of the follow-up commit, the two test classes executed, quickCheck, and three single-change reverts to confirm each new test fails without its production change. Attribution per the ASF Generative Tooling guidance.}

[lhotari]

Non-blocking: three comments that would have saved the next reader two questions

CI is green on 711339c (44/44) and all review threads are resolved, so this is not a merge blocker — take it, take part of it, or push it to a follow-up. I'm raising it because I have unusually direct evidence for it rather than a style opinion.

After reviewing this PR twice and tracing the code with full repository context, I still had to go and answer two questions that the changed code should have answered on its own:

1. Why is the redeliveryMessages.getHash(...) lookup performed at all when isAllowOutOfOrderDelivery() is true? The deleted early return read as a semantic statement ("out-of-order, so nothing to filter"), which makes the new unconditional lookup look like dead work in that mode. It isn't — but nothing at the call site says so.
2. Could private ConcurrentLongLongPairHashMap positionToStickyKeyHash be final? Answering it means symbolically evaluating the two-arm if/else if in add(), noticing the first arm falls through without a return, negating the disjunction in the second, and cross-checking sentinel normalisation in a third file. That's an experiment, not a read — and hashesRefCount being final on the very next line makes the asymmetry look like an oversight rather than a decision.

Worth stressing on (1): the lookup is necessary, and I'm not asking for a code change there. Out-of-order delivery relaxes ordering, not routing — filterAndGroupEntriesForDispatching still calls selector.select(stickyKeyHash) with no mode check, so every replay position still has exactly one owning consumer, and the hash is the only way to learn which one before spending a read. The risk is the opposite one: a future reader concludes the lookup is vestigial and "cleans up" by restoring the early return, silently undoing this PR. That's the failure mode worth a comment.

The suggestion below is comment-only — verified with git diff -U0 | grep -E '^[+-][^+-]' | grep -vE '^[+-][[:space:]]*//' returning nothing, so no executable statement is added, removed or reordered. +13/−1 across two files. Locally: MessageRedeliveryControllerTest 8/8 and PersistentStickyKeyDispatcherMultipleConsumersTest 23/23 pass, ./gradlew quickCheck is clean. Longest added line is 117 chars.

--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
@@ -44,7 +44,12 @@ public class MessageRedeliveryController {
     private final boolean allowOutOfOrderDelivery;
     private final boolean isClassicDispatcher;
     private final ConcurrentBitmapSortedLongPairSet messagesToRedeliver;
+    // Not final: under out-of-order delivery, whether this map is ever needed isn't knowable at construction, since
+    // only a Key_Shared dispatcher records hashes and a plain Shared one never does. add() therefore allocates it on
+    // the first position carrying a real hash. Classic out-of-order never allocates it at all: add() returns at the
+    // isClassicDispatcher branch before reaching ensurePositionToStickyKeyHashMap(), whatever hash is passed.
     private ConcurrentLongLongPairHashMap positionToStickyKeyHash;
+    // Final by contrast: this one is needed exactly when ordering is enforced, which is known at construction.
     private final ConcurrentLongLongHashMap hashesRefCount;

--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -340,6 +340,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
             // instead issue a normal read that waits at the end of the topic for new entries. That would leave
             // deliverable messages stuck in the replay queue or the delayed delivery tracker until an unrelated event
             // (such as a consumer flow request) triggers another read, stalling dispatch (issue #21554).
+            // The former "allowOutOfOrderDelivery ||" escape here was dropped once ReplayPositionFilter started
+            // filtering out-of-order replay positions by available permits; don't re-add one without removing that.
             skipNextReplayToTriggerLookAhead = true;
             // skip backoff delay before reading ahead in the "look ahead" mode to prevent any additional latency
             skipNextBackoff = true;
@@ -589,7 +591,12 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
 
         @Override
         public boolean test(Position position) {
-            // lookup the sticky key hash for the entry at the replay position
+            // Out-of-order delivery relaxes ordering, not routing: filterAndGroupEntriesForDispatching selects the
+            // owning consumer by hash in both modes, so the hash is needed in both. It feeds the permit check below,
+            // which keeps this read's bounded budget (see MessageRedeliveryController#getMessagesToReplayNow) off
+            // positions that dispatch would only push straight back into the replay queue. Under out-of-order
+            // delivery that check and the no-consumer check are the only live rejections here, and it is what made
+            // the "allowOutOfOrderDelivery ||" look-ahead escape removable.
             Long stickyKeyHash = redeliveryMessages.getHash(position.getLedgerId(), position.getEntryId());
             if (stickyKeyHash == null) {
                 // The sticky key hash is missing for delayed messages and positions added through hash-less

Why these three and not more:

• The filter comment is the one that matters. It supplies the missing premise (routing is unconditional), the missing cost model (accepting a position spends a slot of the bounded budget and commits to a ledger read), and the missing fact that only two of the six rejection paths are live under out-of-order delivery — alreadyBlockedHashes is provably empty there (its sole add site is behind !allowOutOfOrderDelivery) and drainingHashesRequired is statically false. Today that has to be re-derived by auditing three separate places.
• The look-ahead comment is the other half of the same argument. The two edits are ~250 lines apart and neither mentions the other, so reviewing either hunk alone reads as an unjustified behaviour change. It also states the coupling as a rule, which is what protects it.
• The field comments answer the final question as a reason, at the declaration.

Optional, and the only executable change I'd suggest — it does mean another CI cycle, so entirely your call whether it is worth disturbing a green run:

-        private final Set<Long> alreadyBlockedHashes = new HashSet<>();
+        private final Set<Long> hashesBlockedForOrdering = new HashSet<>();

There are currently two different things named alreadyBlockedHashes in this one class: this Set<Long> field in ReplayPositionFilter, and the method-local IntOpenHashSet alreadyBlockedHashes in filterAndGroupEntriesForDispatching. They have opposite mode behaviour — the local one is written unconditionally in both modes, this one only when ordering is required. Anyone who reads the dispatch path first (it's the main path, ~140 lines above) carries exactly the wrong mental model into the filter. The rename kills the homonym and puts the ordering-only purpose somewhere it can't rot.

Happy for any of this to become a follow-up instead — the fix itself looks good to me and I'd rather see it merged than perfected.

_{Assisted-by: Claude (Opus 5). The two questions above are genuine — I asked them while re-reviewing, which is what prompted this. The suggested wording was drafted against the full checkout, every factual claim in it re-verified at the source, and the patch built and tested locally as described. Attribution per the ASF Generative Tooling guidance.}

[lhotari]

Approved. Added some non-blocking review comments.