[merlimat]

Supersedes #18227.

Motivation

Problem. Suppose a Shared or Key_Shared subscription's last consumer disconnects while a replay read is in flight, and a new consumer connects before that read completes. The new consumer then gets the replayed entries twice: once from the replay read, and once more when the dispatcher rewinds the cursor. #17163 and #18289 fixed the duplicates reported in #16795 only in the streaming dispatcher, which has since been removed. PersistentDispatcherMultipleConsumers still has the race, and Key_Shared inherits it.

Example. Consumer A, the only consumer, redelivers position P:

1. internalReadMoreEntries starts a replay read of P (havePendingReplayRead = true).

2. A disconnects. clearComponentsAfterRemovedAllConsumers clears redeliveryMessages. cancelPendingRead only cancels normal reads, so the replay read keeps running.

3. B connects. addConsumer sees the pending read and defers the rewind with shouldRewindBeforeReadingOrReplaying = true.

4. The replay read completes. readEntriesComplete discards a read only when it's a normal read:

   if (shouldRewindBeforeReadingOrReplaying && readType == ReadType.Normal) {

   So P is dispatched to B.

5. The next normal read completes. It's discarded, the cursor is rewound to the mark-delete position, and P is read and dispatched again.

Modifications

• readEntriesComplete discards the entries of any read, normal or replay, that completes while the rewind is pending. It calls onReadEnded with the read's own type, so Key_Shared's replay-read bookkeeping stays balanced.
• The rewind now happens in one place, internalReadMoreEntries. While a read that started before the disconnection is pending, that method starts no reads. Once no such read is pending, it rewinds the cursor and clears the flag. Before this change, the first read to end rewound the cursor, so another pending read could still dispatch what the rewind reads again. readEntriesFailed no longer rewinds itself: its retry goes through the same path.

Verifying this change

• PersistentStickyKeyDispatcherMultipleConsumersTest#testReplayReadPendingWhenAllConsumersDisconnectIsNotDispatchedAgain covers both Shared and Key_Shared. It models the cursor's read position and rewind, holds the replay read with a future, and has no sleeps. Without the fix, B receives [1:1, 1:1, 1:2]; with it, [1:1, 1:2].
• PersistentStickyKeyDispatcherMultipleConsumersTest, PersistentDispatcherMultipleConsumersTest, PersistentDispatcherMultipleConsumersReadLimitsTest, SimpleProducerConsumerTest and KeySharedSubscriptionDisabledBrokerCacheTest pass.

[lhotari]

LGTM. Thanks for tracking this race down and for the careful write-up. The fix does what the description says: a read that completes while the rewind is pending is discarded whatever its type, and internalReadMoreEntries is now the only place that rewinds, after no read from before the disconnect is pending. Discarding a replay read doesn't lose anything, because the rewind re-reads everything after the mark-delete position and skips individually acked entries, and onReadEnded(readType) keeps the Key_Shared replay bookkeeping balanced. I couldn't construct a stall: the waiting normal read is cancelled when the last consumer is removed, and a failed read clears its flag before the retry goes through the guarded path. I also reverted the main change locally and the new test fails for both Shared and Key_Shared with [1:1, 1:1, 1:2]; with the change it passes, as do the existing dispatcher test classes.

One non-blocking observation, not introduced here and not something this PR needs to fix: with dispatcherDispatchMessagesInSubscriptionThread enabled, a completed read clears its pending flag before its dispatch is queued, so addConsumer can still rewind while that dispatch is queued, and the new consumer can then receive the entries once from the queued dispatch and again after the rewind. The new test covers the direct-dispatch path only. It may be worth a follow-up issue.

[codelipenghui]

LGTM. The fix discards stale replay reads and defers rewind until pending reads finish, while preserving Key_Shared bookkeeping.

Verified the regression passes for Shared and Key_Shared. With the production fix temporarily reversed, the Shared case fails with duplicate delivery [1:1, 1:1, 1:2]. After restoring the fix, all five selected regression and adjacent lifecycle cases passed. No actionable findings identified within the reviewed scope.