Repository navigation
[5.19.x] [#2583] Duplicate message publish during failover leads to incorrect … - #2714
Merged
mattrpav merged 2 commits intoOct 21, 2026
Merged
Conversation
…rrect queue size metric (apache#2584) * Add tests confirming queueSize accounting for duplicate messages Two end-to-end tests exercise the duplicate paths through the real broker and confirm the messages-counter invariant (messages == 0 after drain, enqueues - dequeues == messages). DuplicateSendStoreRaceQueueSizeTest reproduces the async/sync duplicate send race: an original persistent send from a useAsyncSend producer is counted and cursor-cached while its KahaDB index update is parked on the store executor; a duplicate resend on a different connection takes the sync store path, finds no index row, is assigned a real sequence, passes the -1 duplicate gate, and is counted again. After drain: messages=1, enqueues=2, dequeues=1, failing until the store treats a sync add racing a pending async add of the same MessageId as a duplicate. DuplicateFromStoreEndToEndTest covers both duplicate classes without invoking duplicateFromStore directly. The duplicate SEND is trapped by the cursor cache audit and fully suppressed. The store RE-READ ("cursor got duplicate from store") is a single-counted message read twice whose inflight copy is still consumed and acked; duplicateFromStore must not decrement the messages counter as well, or the count goes negative. A blanket decrement in duplicateFromStore was prototyped and rejected because this test showed messages=-2 for 4 sent and 4 consumed. * Reject sync duplicate add racing a pending async add of the same message With concurrentStoreAndDispatchQueues=true an async add is counted and dispatched from the cursor cache immediately while its journal/index update runs on the store executor. A duplicate send of the same MessageId taking the synchronous addMessage path in that window (failover replay from an async-send producer, arriving on a different connection so the producer audit cannot suppress it) found no row in messageIdIndex, was assigned a real sequence, and evaded the -1 duplicate suppression in Queue.doPendingCursorAdditions, inflating enqueues and the queue messages counter by one for a message that is stored and delivered once. The original's deferred index update then lost the race and was rejected, leaving the count permanently inflated. addMessage now consults asyncTaskMap for a pending async add of the same MessageId before journaling. Sync callers are identified by a null FutureOrSequenceLong (the async task's own store call carries its future), matching the discrimination already used by the IndexAware callback. On a hit the message is marked with the -1 duplicate sequence and skipped entirely: no journal write, no pendingAdds entry, and the existing cursor suppression keeps it uncounted. (cherry picked from commit e4fd919)
mattrpav
force-pushed
the
amq-dupe-size-failover-5.19.x
branch
from
October 21, 2026 12:16
d9c9c4c to
7511a17
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
…queue size metric (#2584)
Two end-to-end tests exercise the duplicate paths through the real broker and confirm the messages-counter invariant (messages == 0 after drain, enqueues - dequeues == messages).
DuplicateSendStoreRaceQueueSizeTest reproduces the async/sync duplicate send race: an original persistent send from a useAsyncSend producer is counted and cursor-cached while its KahaDB index update is parked on the store executor; a duplicate resend on a different connection takes the sync store path, finds no index row, is assigned a real sequence, passes the -1 duplicate gate, and is counted again. After drain: messages=1, enqueues=2, dequeues=1, failing until the store treats a sync add racing a pending async add of the same MessageId as a duplicate.
DuplicateFromStoreEndToEndTest covers both duplicate classes without invoking duplicateFromStore directly. The duplicate SEND is trapped by the cursor cache audit and fully suppressed. The store RE-READ ("cursor got duplicate from store") is a single-counted message read twice whose inflight copy is still consumed and acked; duplicateFromStore must not decrement the messages counter as well, or the count goes negative. A blanket decrement in duplicateFromStore was prototyped and rejected because this test showed messages=-2 for 4 sent and 4 consumed.
With concurrentStoreAndDispatchQueues=true an async add is counted and dispatched from the cursor cache immediately while its journal/index update runs on the store executor. A duplicate send of the same MessageId taking the synchronous addMessage path in that window (failover replay from an async-send producer, arriving on a different connection so the producer audit cannot suppress it) found no row in messageIdIndex, was assigned a real sequence, and evaded the -1 duplicate suppression in Queue.doPendingCursorAdditions, inflating enqueues and the queue messages counter by one for a message that is stored and delivered once. The original's deferred index update then lost the race and was rejected, leaving the count permanently inflated.
addMessage now consults asyncTaskMap for a pending async add of the same MessageId before journaling. Sync callers are identified by a null FutureOrSequenceLong (the async task's own store call carries its future), matching the discrimination already used by the IndexAware callback. On a hit the message is marked with the -1 duplicate sequence and skipped entirely: no journal write, no pendingAdds entry, and the existing cursor suppression keeps it uncounted.
(cherry picked from commit e4fd919)