mattrpav opened a new pull request, #2717:
URL: https://github.com/apache/activemq/pull/2717
…queue size metric (#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 e4fd919b335299716e00d1b6ff096bb0d50deb17)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
