mattrpav opened a new pull request, #2715: URL: https://github.com/apache/activemq/pull/2715
…turely leading to incorrect queue size (#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 d0c5ff05dec1ec1fc2e7b29d675bbca7288bd294) -- 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
