Repository navigation
Conversation
… prematurely leading to incorrect queue size (apache#2585) * Add test proving incorrect queueSize when a post-commit synchronization throws For a transacted acknowledgment two synchronizations run in order on commit: the subscription's (dequeue bookkeeping, then nodeDest.wakeup() and dispatchPending()) followed by the queue's, whose afterCommit calls dropMessage(), the only decrement of the queue messages counter. Transaction.fireAfterCommit() iterates the synchronization list with no per-synchronization exception isolation, so the first throw aborts the loop. The KahaDB store commit is already durable at that point (the index ack applies before the postCommit runnable runs), so an exception from the subscription synchronization (dispatchPending() hitting a transient cursor/store error, which with useCache=false touches the store on every commit) permanently loses the decrement: the message is removed from the store and not redeliverable, but the messages counter stays +1 with the queue drained, browse empty and inflight 0. The client sees "STORE COMMIT FAILED", the forensic log marker for this failure. The test swaps the subscription's pending cursor for one whose reset() throws when invoked from within afterCommit (the ack path also calls dispatchPending and must not be poisoned), commits, verifies the store is empty and nothing is redeliverable, and asserts the messages counter returns to 0. Fails with messages=1, dequeues=0 until fireAfterCommit isolates synchronization failures. * Isolate synchronization failures in fireAfterCommit and fireAfterRollback The transaction outcome is already durable when the afterCommit (and afterRollback) synchronizations run, but fireAfterCommit iterated the synchronization list with no per-synchronization exception isolation: the first throw aborted the loop and silently skipped every remaining synchronization. For a transacted queue acknowledgment the synchronizations run in order [subscription, queue]. The subscription's afterCommit ends with nodeDest.wakeup() and dispatchPending(), throw-capable calls that touch the store cursor on every commit when useCache=false. One transient failure there skipped the queue synchronization, whose afterCommit calls dropMessage(), the only site that decrements the queue messages counter and increments destination dequeues. The store ack was already applied, so the counter inflated by one permanently: queue drained, browse empty, inflight zero, "STORE COMMIT FAILED" in the log. The TransactedProduceWithThrowingAckQueueSizeTest.java test confirms the produced message is not stranded: it is delivered exactly once, the store drains to empty, and queueSize settles to 0. It guards against a regression where the rollback would leave a durably committed message out of the cursor until a broker restart. * Catch Exception over Throwable (cherry picked from commit d0c5ff0)
mattrpav
force-pushed
the
amq-dupe-size-tx-commit-5.19.x
branch
from
October 21, 2026 11:34
44dcd88 to
d8f87ef
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.
…turely leading to incorrect queue size (#2585)
For a transacted acknowledgment two synchronizations run in order on commit: the subscription's (dequeue bookkeeping, then nodeDest.wakeup() and dispatchPending()) followed by the queue's, whose afterCommit calls dropMessage(), the only decrement of the queue messages counter.
Transaction.fireAfterCommit() iterates the synchronization list with no per-synchronization exception isolation, so the first throw aborts the loop. The KahaDB store commit is already durable at that point (the index ack applies before the postCommit runnable runs), so an exception from the subscription synchronization (dispatchPending() hitting a transient cursor/store error, which with useCache=false touches the store on every commit) permanently loses the decrement: the message is removed from the store and not redeliverable, but the messages counter stays +1 with the queue drained, browse empty and inflight 0. The client sees "STORE COMMIT FAILED", the forensic log marker for this failure.
The test swaps the subscription's pending cursor for one whose reset() throws when invoked from within afterCommit (the ack path also calls dispatchPending and must not be poisoned), commits, verifies the store is empty and nothing is redeliverable, and asserts the messages counter returns to 0. Fails with messages=1, dequeues=0 until fireAfterCommit isolates synchronization failures.
The transaction outcome is already durable when the afterCommit (and afterRollback) synchronizations run, but fireAfterCommit iterated the synchronization list with no per-synchronization exception isolation: the first throw aborted the loop and silently skipped every remaining synchronization.
For a transacted queue acknowledgment the synchronizations run in order [subscription, queue]. The subscription's afterCommit ends with nodeDest.wakeup() and dispatchPending(), throw-capable calls that touch the store cursor on every commit when useCache=false. One transient failure there skipped the queue synchronization, whose afterCommit calls dropMessage(), the only site that decrements the queue messages counter and increments destination dequeues. The store ack was already applied, so the counter inflated by one permanently: queue drained, browse empty, inflight zero, "STORE COMMIT FAILED" in the log.
The TransactedProduceWithThrowingAckQueueSizeTest.java test confirms the produced message is not stranded: it is delivered
exactly once, the store drains to empty, and queueSize settles to 0. It guards against a regression where the rollback would leave a durably committed message out of the cursor until a broker restart.
(cherry picked from commit d0c5ff0)