You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ma...@apache.org on 2026/10/21 17:31:30 UTC
(activemq) branch activemq-5.19.x updated: [5.19.x] [#2583] Transacted fire-after-commit first exception exits loop prema… (#2715)
This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch activemq-5.19.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.19.x by this push:
new 9351c98117 [5.19.x] [#2583] Transacted fire-after-commit first exception exits loop prema… (#2715)
9351c98117 is described below
commit 9351c9811753e5101f57e3659d90cec9f370e5a4
Author: Matt Pavlovich <ma...@apache.org>
AuthorDate: Wed Oct 21 13:31:24 2026 -0400
[5.19.x] [#2583] Transacted fire-after-commit first exception exits loop prema… (#2715)
* [#2583] Transacted fire-after-commit first exception exits loop prematurely 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)
* [#2583] 5.19.x backport compile fixes for unit tests
---
.../apache/activemq/transaction/Transaction.java | 31 ++-
.../bugs/TransactedAckPostCommitQueueSizeTest.java | 247 +++++++++++++++++++++
...nsactedProduceWithThrowingAckQueueSizeTest.java | 235 ++++++++++++++++++++
3 files changed, 511 insertions(+), 2 deletions(-)
diff --git a/activemq-broker/src/main/java/org/apache/activemq/transaction/Transaction.java b/activemq-broker/src/main/java/org/apache/activemq/transaction/Transaction.java
index 09334803cd..03312771ab 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/transaction/Transaction.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/transaction/Transaction.java
@@ -134,20 +134,47 @@ public abstract class Transaction {
}
protected void fireAfterCommit() throws Exception {
+ // The transaction outcome is already durable when afterCommit runs, so a
+ // failing synchronization must not prevent the remaining ones from
+ // executing (e.g. the destination synchronization that maintains the
+ // destination statistics) - run them all and rethrow the first failure.
+ // Only Exceptions are isolated; a fatal Error aborts the chain immediately.
+ Exception firstException = null;
synchronized(synchronizations) {
for (Iterator<Synchronization> iter = synchronizations.iterator(); iter.hasNext();) {
Synchronization s = iter.next();
- s.afterCommit();
+ try {
+ s.afterCommit();
+ } catch (Exception e) {
+ getLog().warn("Exception from afterCommit for {} of {}", s, this, e);
+ if (firstException == null) {
+ firstException = e;
+ }
+ }
}
}
+ if (firstException != null) {
+ throw firstException;
+ }
}
public void fireAfterRollback() throws Exception {
synchronized(synchronizations) {
Collections.reverse(synchronizations);
+ Exception firstException = null;
for (Iterator<Synchronization> iter = synchronizations.iterator(); iter.hasNext();) {
Synchronization s = iter.next();
- s.afterRollback();
+ try {
+ s.afterRollback();
+ } catch (Exception e) {
+ getLog().warn("Exception from afterRollback for {} of {}", s, this, e);
+ if (firstException == null) {
+ firstException = e;
+ }
+ }
+ }
+ if (firstException != null) {
+ throw firstException;
}
}
}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedAckPostCommitQueueSizeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedAckPostCommitQueueSizeTest.java
new file mode 100644
index 0000000000..f45fca216d
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedAckPostCommitQueueSizeTest.java
@@ -0,0 +1,247 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.PrefetchSubscription;
+import org.apache.activemq.broker.region.Queue;
+import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.IOHelper;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reproduces queue size defect D1: an exception thrown from an
+ * earlier transaction Synchronization skips the queue's dropMessage
+ * synchronization (docs/bugs/queuesize-deep-dive.md).
+ *
+ * For a transacted acknowledgment two synchronizations are registered in
+ * order: the subscription's (PrefetchSubscription.registerRemoveSync: sub
+ * dequeue counters, dispatched-list removal, then nodeDest.wakeup() and
+ * dispatchPending()) followed by the queue's (Queue.removeMessage, whose
+ * afterCommit calls dropMessage(), the only decrement of the queue messages
+ * counter).
+ *
+ * Transaction.fireAfterCommit() iterates the synchronization list with no
+ * per-synchronization exception isolation: the first throw aborts the loop.
+ * The KahaDB store commit is already durable at that point (the index ack is
+ * applied before the postCommit runnable runs), so a throw from the
+ * subscription synchronization (for example dispatchPending() hitting a
+ * transient cursor/store error, which with useCache=false does store work on
+ * every commit) permanently loses the decrement:
+ *
+ * message removed from the store and never re-deliverable,
+ * but messages counter stays +1: queue drained, browse empty, inflight 0.
+ *
+ * The test injects a one-shot RuntimeException from the subscription's
+ * pending cursor reset() (the first call dispatchPending makes), commits,
+ * and asserts the counter returns to 0.
+ */
+public class TransactedAckPostCommitQueueSizeTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TransactedAckPostCommitQueueSizeTest.class);
+ private static final String QUEUE_NAME = "TEST.TX.POST.COMMIT.QUEUE.SIZE";
+
+ private BrokerService broker;
+ private Connection connection;
+ private File dataDir;
+
+ /**
+ * Pending cursor that throws from reset() while armed, but ONLY when
+ * invoked from within a transaction Synchronization's afterCommit. The
+ * ack-processing path also calls dispatchPending() and must not be
+ * poisoned (that would mark the transaction rollback-only before the
+ * store commit, a different scenario).
+ */
+ static class FailingResetPendingCursor extends VMPendingMessageCursor {
+ final AtomicBoolean armed = new AtomicBoolean(false);
+ final AtomicInteger throwCount = new AtomicInteger(0);
+
+ FailingResetPendingCursor(boolean prioritizedMessages) {
+ super(prioritizedMessages);
+ }
+
+ @Override
+ public synchronized void reset() {
+ if (armed.get() && calledFromAfterCommit()) {
+ throwCount.incrementAndGet();
+ throw new RuntimeException("injected: transient cursor failure during post-commit dispatchPending");
+ }
+ super.reset();
+ }
+
+ private boolean calledFromAfterCommit() {
+ for (var frame : Thread.currentThread().getStackTrace()) {
+ if ("afterCommit".equals(frame.getMethodName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ var baseDir = new File(IOHelper.getDefaultDataDirectory());
+ Files.createDirectories(baseDir.toPath());
+ dataDir = Files.createTempDirectory(baseDir.toPath(), "TxPostCommitQueueSize-").toFile();
+ dataDir.deleteOnExit();
+
+ broker = new BrokerService();
+ broker.setDataDirectoryFile(dataDir);
+ broker.setUseJmx(false);
+ broker.setDeleteAllMessagesOnStartup(true);
+ broker.getSystemUsage().getMemoryUsage().setLimit(64 * 1024 * 1024);
+
+ var pa = new KahaDBPersistenceAdapter();
+ pa.setDirectory(new File(dataDir, "kahadb"));
+ broker.setPersistenceAdapter(pa);
+
+ // Production profile: no cursor cache, small page size
+ var policyMap = new PolicyMap();
+ var entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setUseCache(false);
+ entry.setMaxPageSize(60);
+ policyMap.setDefaultEntry(entry);
+ broker.setDestinationPolicy(policyMap);
+
+ broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ var factory = new ActiveMQConnectionFactory(
+ broker.getTransportConnectors().get(0).getConnectUri());
+ connection = factory.createConnection();
+ connection.start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (connection != null) {
+ connection.close();
+ }
+ if (broker != null) {
+ broker.deleteAllMessages();
+ broker.stop();
+ broker.waitUntilStopped();
+ }
+ }
+
+ @Test(timeout = 60_000)
+ public void testPostCommitExceptionDoesNotLoseDecrement() throws Exception {
+ var dest = new ActiveMQQueue(QUEUE_NAME);
+
+ // Send one persistent message
+ try (var producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var producer = producerSession.createProducer(dest)) {
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(producerSession.createTextMessage("m1"));
+ }
+
+ var queue = (Queue) broker.getDestination(dest);
+ var stats = queue.getDestinationStatistics();
+ assertTrue("message should be counted",
+ Wait.waitFor(() -> stats.getMessages().getCount() == 1, 5000, 10));
+
+ // Transacted consume (DMLC style: 1 message per commit)
+ try (var txSession = connection.createSession(true, Session.SESSION_TRANSACTED);
+ var consumer = txSession.createConsumer(dest)) {
+ Message received = consumer.receive(5000);
+ assertNotNull("message should be received", received);
+
+ // Swap the subscription's pending cursor for one that throws from
+ // reset(), the first call dispatchPending() makes inside the
+ // subscription synchronization's afterCommit.
+ var sub = (PrefetchSubscription) queue.getConsumers().get(0);
+ var failingCursor = new FailingResetPendingCursor(false);
+ sub.setPending(failingCursor);
+ failingCursor.armed.set(true);
+
+ try {
+ txSession.commit();
+ LOG.info("commit() returned normally");
+ } catch (JMSException expected) {
+ LOG.info("commit() threw (expected while the post-commit chain aborts): {}",
+ expected.toString());
+ } finally {
+ failingCursor.armed.set(false);
+ }
+
+ assertTrue("injected failure should have fired during afterCommit",
+ failingCursor.throwCount.get() >= 1);
+ }
+
+ // The store ack was durable before the synchronizations ran: the
+ // message must be gone from the store and not redeliverable.
+ assertTrue("store should be empty after commit",
+ Wait.waitFor(() -> {
+ try {
+ return queue.getMessageStore().getMessageCount() == 0;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 5000, 100));
+
+ try (var verifySession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var verifyConsumer = verifySession.createConsumer(dest)) {
+ assertNull("no message should be redeliverable", verifyConsumer.receive(2000));
+ }
+
+ var settled = Wait.waitFor(() -> stats.getMessages().getCount() == 0, 5000, 100);
+
+ LOG.info("Final stats: messages={}, enqueues={}, dequeues={}, inflight={}",
+ stats.getMessages().getCount(), stats.getEnqueues().getCount(),
+ stats.getDequeues().getCount(), stats.getInflight().getCount());
+
+ assertTrue("INCORRECT QUEUE SIZE (D1): message committed+consumed (store empty, nothing " +
+ "redeliverable, inflight=" + stats.getInflight().getCount() + ") but messages " +
+ "counter is " + stats.getMessages().getCount() + "; an exception from the " +
+ "subscription's afterCommit synchronization skipped the queue's dropMessage " +
+ "synchronization in Transaction.fireAfterCommit",
+ settled);
+
+ assertEquals("dequeues should reflect the consumed message",
+ 1, stats.getDequeues().getCount());
+ }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedProduceWithThrowingAckQueueSizeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedProduceWithThrowingAckQueueSizeTest.java
new file mode 100644
index 0000000000..8c0b077374
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedProduceWithThrowingAckQueueSizeTest.java
@@ -0,0 +1,235 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.PrefetchSubscription;
+import org.apache.activemq.broker.region.Queue;
+import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.IOHelper;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Checks the side effect of the fireAfterCommit isolation fix when a single
+ * transaction both produces and consumes: after the fix, when a post-commit
+ * synchronization throws, every synchronization runs afterCommit and then the
+ * triggered rollback runs afterRollback. For a produce in that same
+ * transaction the queue's CursorAddSync gets afterCommit (pages the message in,
+ * counts it) and then afterRollback (rollbackPendingCursorAdditions) - while
+ * the message is already durably committed in the store.
+ *
+ * The transaction here consumes M1 (its ack's afterCommit is rigged to throw)
+ * and produces M2. The concern is that M2 is left durably in the store but
+ * pulled back out of the in-memory cursor, stranded until a restart. The test
+ * confirms this is not the case: the store must drain to empty and queueSize
+ * must settle to 0, meaning M2 is still deliverable.
+ */
+public class TransactedProduceWithThrowingAckQueueSizeTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TransactedProduceWithThrowingAckQueueSizeTest.class);
+ private static final String QUEUE_NAME = "TEST.TX.PRODUCE.THROWING.ACK";
+
+ private BrokerService broker;
+ private Connection connection;
+ private File dataDir;
+
+ /** Pending cursor that throws from reset() while armed, but only inside afterCommit. */
+ static class FailingResetPendingCursor extends VMPendingMessageCursor {
+ final AtomicBoolean armed = new AtomicBoolean(false);
+ final AtomicInteger throwCount = new AtomicInteger(0);
+
+ FailingResetPendingCursor(boolean prioritizedMessages) {
+ super(prioritizedMessages);
+ }
+
+ @Override
+ public synchronized void reset() {
+ if (armed.get() && calledFromAfterCommit()) {
+ throwCount.incrementAndGet();
+ throw new RuntimeException("injected: transient cursor failure during post-commit dispatchPending");
+ }
+ super.reset();
+ }
+
+ private boolean calledFromAfterCommit() {
+ for (var frame : Thread.currentThread().getStackTrace()) {
+ if ("afterCommit".equals(frame.getMethodName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ var baseDir = new File(IOHelper.getDefaultDataDirectory());
+ Files.createDirectories(baseDir.toPath());
+ dataDir = Files.createTempDirectory(baseDir.toPath(), "TxProduceThrowingAck-").toFile();
+ dataDir.deleteOnExit();
+
+ broker = new BrokerService();
+ broker.setDataDirectoryFile(dataDir);
+ broker.setUseJmx(false);
+ broker.setDeleteAllMessagesOnStartup(true);
+ broker.getSystemUsage().getMemoryUsage().setLimit(64 * 1024 * 1024);
+
+ var pa = new KahaDBPersistenceAdapter();
+ pa.setDirectory(new File(dataDir, "kahadb"));
+ broker.setPersistenceAdapter(pa);
+
+ // Production profile: no cursor cache, small page size
+ var policyMap = new PolicyMap();
+ var entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setUseCache(false);
+ entry.setMaxPageSize(60);
+ policyMap.setDefaultEntry(entry);
+ broker.setDestinationPolicy(policyMap);
+
+ broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ var factory = new ActiveMQConnectionFactory(
+ broker.getTransportConnectors().get(0).getConnectUri());
+ connection = factory.createConnection();
+ connection.start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (connection != null) {
+ connection.close();
+ }
+ if (broker != null) {
+ broker.deleteAllMessages();
+ broker.stop();
+ broker.waitUntilStopped();
+ }
+ }
+
+ @Test(timeout = 60_000)
+ public void testProducedMessageNotStrandedWhenAckAfterCommitThrows() throws Exception {
+ var dest = new ActiveMQQueue(QUEUE_NAME);
+
+ // Seed M1 (committed, outside any transaction) so the transacted session
+ // has a message to consume.
+ try (var seedSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var seedProducer = seedSession.createProducer(dest)) {
+ seedProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ seedProducer.send(seedSession.createTextMessage("M1-consumed"));
+ }
+
+ var queue = (Queue) broker.getDestination(dest);
+ var stats = queue.getDestinationStatistics();
+ assertTrue("seed message should be counted",
+ Wait.waitFor(() -> stats.getMessages().getCount() == 1, 5000, 10));
+
+ // One transaction that produces M2 and consumes M1. Producing M2 adds a
+ // CursorAddSync; committing the consume adds the ack synchronizations.
+ try (var txSession = connection.createSession(true, Session.SESSION_TRANSACTED)) {
+ try (var producer = txSession.createProducer(dest)) {
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(txSession.createTextMessage("M2-produced"));
+ }
+
+ var consumer = txSession.createConsumer(dest);
+ var m1 = consumer.receive(5000);
+ assertNotNull("M1 should be received in the transaction", m1);
+ assertEquals("M1-consumed", ((TextMessage) m1).getText());
+
+ // Rig the ack's afterCommit to throw, as in the D1 reproduction.
+ var sub = (PrefetchSubscription) queue.getConsumers().get(0);
+ var failingCursor = new FailingResetPendingCursor(false);
+ sub.setPending(failingCursor);
+ failingCursor.armed.set(true);
+
+ try {
+ txSession.commit();
+ LOG.info("commit() returned normally");
+ } catch (JMSException expected) {
+ LOG.info("commit() threw (post-commit chain aborted, rollback triggered): {}",
+ expected.toString());
+ } finally {
+ failingCursor.armed.set(false);
+ }
+
+ assertTrue("injected afterCommit failure should have fired",
+ failingCursor.throwCount.get() >= 1);
+ consumer.close();
+ }
+
+ // Drain whatever is deliverable with a fresh consumer.
+ var received = new ArrayList<String>();
+ try (var drainSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var drainConsumer = drainSession.createConsumer(dest)) {
+ javax.jms.Message m;
+ while ((m = drainConsumer.receive(2000)) != null) {
+ received.add(((TextMessage) m).getText());
+ }
+ }
+
+ long storeCount = queue.getMessageStore().getMessageCount();
+ LOG.info("After drain: received={}, messages={}, store={}, enqueues={}, dequeues={}",
+ received, stats.getMessages().getCount(), storeCount,
+ stats.getEnqueues().getCount(), stats.getDequeues().getCount());
+
+ // The produced M2 was durably committed to the store. It must not be
+ // stranded there: the store must drain to empty and queueSize settle to
+ // 0. A stranded M2 leaves store count and queueSize stuck at 1.
+ assertTrue("LOST/STRANDED MESSAGE: the produced message is durably in the store but was " +
+ "pulled out of the cursor by CursorAddSync.afterRollback and never redelivered " +
+ "(store count " + queue.getMessageStore().getMessageCount() + ", received " + received + ")",
+ Wait.waitFor(() -> {
+ try {
+ return queue.getMessageStore().getMessageCount() == 0;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 8000, 100));
+
+ assertTrue("queueSize must settle to 0",
+ Wait.waitFor(() -> stats.getMessages().getCount() == 0, 5000, 100));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact