[fix][broker] Fix Key_Shared delivery stall when look-ahead triggers at the end of the topic

[lhotari]

Fixes #21554

Motivation

KeySharedSubscriptionTest.testContinueDispatchMessagesWhenMessageDelayed has been flaky for a long time (#21554), failing with assertions such as expected [80] but found [78]. Analysis of the test-report artifact of a recent CI failure (both attempts of the test failed in this run) and a local reproduction with DEBUG logging showed that this is a genuine dispatch stall in the PIP-379 Key_Shared dispatcher, not a test-side problem.

The stall arises from the following sequence:

1. A normal read is parked at the end of the topic waiting for new entries (havePendingRead=true). This is the normal state for a caught-up subscription.
2. Delayed messages become due. getMessagesToReplayNow pulls them from the delayed delivery tracker in permit-limited batches, so some already-due messages remain in the tracker. InMemoryDelayedDeliveryTracker.updateTimer intentionally doesn't arm a timer for already-due messages, on the assumption that a subsequent readMoreEntries call will pull them.
3. One replay batch happens to contain only messages hashed to a consumer without available permits (in the test topology: the consumer whose receiver queue is full). The whole batch gets discarded at dispatch, and since the other consumer has available permits, "look ahead" is engaged by setting skipNextReplayToTriggerLookAhead=true.
4. The follow-up readMoreEntries call consumes the flag and skips both the replay queue and the delayed-tracker pull — and because the cursor has no more entries, it just finds the parked pending read and gives up ("Cannot schedule next read until previous one is done"). The flag has consumed the only pending trigger.
5. No triggers remain: concurrent consumer flow triggers were dropped while the replay read was in flight ("Skipping replay while awaiting previous read to complete"), acknowledgments don't advance the mark-delete position (the other consumer holds an unacked backlog, so markDeletePositionMoveForward never fires), the delayed tracker has no timer armed, and checkAndUnblockIfStuck doesn't consider the dispatcher stuck because a (parked) pending read exists.

Dispatch then stalls until an unrelated event triggers a read. In the test, that's the blocked consumer's flow request 30 seconds later — by which time the test has already given up on the first consumer, so the stalled messages hashed to it are delivered into a receiver queue nobody reads anymore and count as missing.

DEBUG-log excerpt from a local reproduction (1 of 20 runs, masked by the test retry analyzer) showing the stall onset — the 5-entry replay batch discarded and every remaining trigger dropped, followed by 30 seconds of silence:

22:57:58,185 InMemoryDelayedDeliveryTracker - Get scheduled messages {messagesCount=5}
22:57:58,185 PersistentDispatcherMultipleConsumers - Schedule replay of messages for consumers {messagesToReplay=5}
22:57:58,186 PersistentDispatcherMultipleConsumers - Skipping replay while awaiting previous read to complete
22:57:58,186 PersistentDispatcherMultipleConsumers - Adding message to replay for: hash {entryId=30 ... entryId=34}
22:57:58,186 PersistentDispatcherMultipleConsumers - Reschedule message read {readAfterMs=0}
22:57:58,186 PersistentDispatcherMultipleConsumers - Cannot schedule next read until previous one is done
        (30 seconds of silence; recovery only via the other consumer's flow request)
22:58:28,197 InMemoryDelayedDeliveryTracker - Get scheduled messages

Modifications

• When ordered delivery is required, engage look-ahead in PersistentStickyKeyDispatcherMultipleConsumers.trySendMessagesToConsumers only when the cursor has more entries to read. When the cursor is caught up there is nothing to look ahead at, so skipping the next replay could only cause the stall described above. With the change, the follow-up read replays the replay queue and pulls due messages from the delayed delivery tracker instead of parking a normal read at the end of the topic. The ReplayPositionFilter keeps this loop-free for ordered delivery: once a discarded message's hash is recorded in the replay queue, subsequent replay selections exclude messages for consumers without available permits.
• When out-of-order delivery is allowed, keep engaging look-ahead unconditionally (the previous behavior). In that mode the replay queue doesn't track sticky key hashes, so the replay position filter cannot exclude messages for consumers without available permits, and without the look-ahead cut-off the dispatcher would repeatedly re-read and discard the same undispatchable messages. This was caught by KeySharedSubscriptionTest.testRecentJoinedPosWillNotStuckOtherConsumer[PIP379, allowOutOfOrder=true] in the first CI run of this PR (replay read count 12229 vs the asserted maximum of 200, reproduced locally with 32662).
• Add a regression test, PersistentStickyKeyDispatcherMultipleConsumersTest.testLookAheadNotEngagedWhenCursorHasNoMoreEntries, that deterministically reproduces the stall shape at the dispatcher level. It fails without the production change and passes with it.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• PersistentStickyKeyDispatcherMultipleConsumersTest.testLookAheadNotEngagedWhenCursorHasNoMoreEntries fails without the production change (the replayed message for the consumer with available permits is never dispatched) and passes with it.
• Verified KeySharedSubscriptionTest.testContinueDispatchMessagesWhenMessageDelayed locally with invocationCount = 20 for the PIP379 implementation: without the fix, 1 of 20 runs hit the stall (a ~70 s run where 5 of 80 messages were delivered only after a 30 s stall, masked by the retry analyzer); with the fix, 20 of 20 runs completed in ≤ 41 s with all 80 messages delivered.
• Verified KeySharedSubscriptionTest.testRecentJoinedPosWillNotStuckOtherConsumer locally: with the end-of-topic guard applied to out-of-order delivery as well, the [PIP379, true] variant hit a repeated read-and-discard loop (replay read count 32662); with look-ahead kept unconditional for out-of-order delivery, all four variants pass with replay read counts of 50/7/56/12. The full KeySharedSubscriptionTest class passes locally.

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