This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit b3d24ca5206de0ab58639feb2d7f99aa70d52a5b Author: Aurélien Pupier <[email protected]> AuthorDate: Thu Nov 5 12:12:14 2026 +0100 CAMEL-26631: camel-sjms - Fix flaky BatchConsumerReconnectTest BatchConsumerReconnectTest failed intermittently in different tests (batch sizes of 3 instead of 5 or 4, unexpected mock message counts), and the Surefire reruns of a failed test then always failed with "AMQ219007: Cannot connect to server(s)". Causes and fixes: - testReconnectSessionTransacted and testReconnectClientAcknowledge sent 2 messages and immediately simulated a connection failure, without waiting for the batch consumer to receive them. Messages not yet received were not part of the buffered batch, so the first batch after the reconnect had fewer messages than expected. Wait until both messages have been received before triggering the failure, as already done for testReconnectAutoAcknowledge (CAMEL-26369). - The session-tx and client-ack routes used a 1000 ms batch interval, so a batch could be completed by the timer instead of by size, depending on how long the reconnection took on a slow machine. Use a 60 s interval for these routes so batches are only completed by size. As a consequence testReconnectSessionTransacted now sends 8 messages after the reconnect and expects two complete batches of 5, as an incomplete batch can no longer be flushed by the timer. The auto-ack route keeps the 1000 ms interval as its test relies on it for an incomplete batch. - AbstractArtemisEmbeddedService.initialize() reused the stopped EmbeddedActiveMQ (and its ActiveMQServer and in-VM acceptor) when the service was started again, which is what happens when Surefire reruns a failed test in the same JVM (afterAll stops the broker, beforeAll initializes it again). A new broker id, hence a new vm:// address, is computed on each initialization but the stopped server still had the previous acceptor, so nothing was listening on the new address. Create a new EmbeddedActiveMQ and Configuration when the previous server has been stopped. How it was verified: - Making a test fail on its first attempt (temporary, not committed) and running with -Dsurefire.rerunFailingTestsCount=2: before the fix the reruns failed with AMQ219007, after the fix the rerun passes (Flakes: 1). - BatchConsumerReconnectTest and BatchConsumerStopStartTest pass. - The original race conditions could not be reproduced locally. Co-authored-by: IBM Bob 2.2.1 Signed-off-by: Aurélien Pupier <[email protected]> --- .../sjms/batch/BatchConsumerReconnectTest.java | 25 ++++++++++++++++------ .../services/AbstractArtemisEmbeddedService.java | 14 +++++++++--- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerReconnectTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerReconnectTest.java index e022639c519a..e38ed0c29786 100644 --- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerReconnectTest.java +++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/BatchConsumerReconnectTest.java @@ -47,6 +47,7 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { private static final String ROUTE_ID_CLIENT_ACK = "client-ack"; private static final String ROUTE_ID_AUTO_ACK = "auto-ack"; private static final String SJMS_CONNECTION_FACTORY = "countingConnectionFactory"; + private static final int LONG_BATCH_INTERVAL = 60000; private BatchTestHelper.TrackedConnectionFactory trackedConnectionFactory; @@ -115,6 +116,10 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { sendMessagesWithText(template, endpointUri, 2, "Before!"); + // make sure both messages are buffered by the consumer before simulating the failure + await().atMost(10, TimeUnit.SECONDS) + .until(() -> trackedConnectionFactory.messages(queueName) == 2); + int connectionsBefore = trackedConnectionFactory.getCreateCount(); triggerConnectionFailure(context, ROUTE_ID_SESSION_TX); await().atMost(30, TimeUnit.SECONDS) @@ -123,9 +128,10 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { mock.assertIsSatisfied(); mock.reset(); - // we want to test with 1 complete (incl. the messages prior to reconnect) and 1 incomplete batch + // we want to test with 2 complete batches (the first one incl. the messages prior to reconnect); + // the batch interval is very long so that batches are only completed by size, not by timing mock.expectedMessageCount(2); - sendMessagesWithText(template, endpointUri, 7, "After!"); + sendMessagesWithText(template, endpointUri, 8, "After!"); mock.assertIsSatisfied(); // first batch @@ -133,8 +139,9 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { assertEquals(List.of("Before!", "Before!", "After!", "After!", "After!"), getBatchBodiesAsString(mock.getExchanges().get(0))); // second batch - assertEquals(4, getBatchExchanges(mock.getExchanges().get(1)).size()); - assertEquals(List.of("After!", "After!", "After!", "After!"), getBatchBodiesAsString(mock.getExchanges().get(1))); + assertEquals(5, getBatchExchanges(mock.getExchanges().get(1)).size()); + assertEquals(List.of("After!", "After!", "After!", "After!", "After!"), + getBatchBodiesAsString(mock.getExchanges().get(1))); } @Test @@ -147,6 +154,10 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { sendMessagesWithText(template, endpointUri, 2, "Before!"); + // make sure both messages are buffered by the consumer before simulating the failure + await().atMost(10, TimeUnit.SECONDS) + .until(() -> trackedConnectionFactory.messages(queueName) == 2); + int connectionsBefore = trackedConnectionFactory.getCreateCount(); triggerConnectionFailure(context, ROUTE_ID_CLIENT_ACK); await().atMost(30, TimeUnit.SECONDS) @@ -165,11 +176,13 @@ public class BatchConsumerReconnectTest extends JmsExclusiveTestSupport { } protected RoutesBuilder[] createRouteBuilders() { + // the session-tx and client-ack routes use a very long batch interval, so that the batches are only + // completed by size and the tests do not depend on timing (slow CI, time taken by the reconnection) return new RoutesBuilder[] { - createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX, 5, 1000, + createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_SESSION_TX, 5, LONG_BATCH_INTERVAL, true, null, 1, new BatchTestHelper.DoNothingProcessor(), "#" + SJMS_CONNECTION_FACTORY), - createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_CLIENT_ACK, 5, 1000, + createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_CLIENT_ACK, 5, LONG_BATCH_INTERVAL, false, "CLIENT_ACKNOWLEDGE", 1, new BatchTestHelper.DoNothingProcessor(), "#" + SJMS_CONNECTION_FACTORY), createBatchRoute(QUEUE_NAME_TEMPLATE, ROUTE_ID_AUTO_ACK, 5, 1000, diff --git a/test-infra/camel-test-infra-artemis/src/main/java/org/apache/camel/test/infra/artemis/services/AbstractArtemisEmbeddedService.java b/test-infra/camel-test-infra-artemis/src/main/java/org/apache/camel/test/infra/artemis/services/AbstractArtemisEmbeddedService.java index 6c23fce4760d..e545dc861786 100644 --- a/test-infra/camel-test-infra-artemis/src/main/java/org/apache/camel/test/infra/artemis/services/AbstractArtemisEmbeddedService.java +++ b/test-infra/camel-test-infra-artemis/src/main/java/org/apache/camel/test/infra/artemis/services/AbstractArtemisEmbeddedService.java @@ -45,8 +45,8 @@ public abstract class AbstractArtemisEmbeddedService implements ArtemisInfraServ protected static final Logger LOG = LoggerFactory.getLogger(AbstractArtemisEmbeddedService.class); private static final AtomicInteger BROKER_COUNT = new AtomicInteger(); - protected final EmbeddedActiveMQ embeddedBrokerService; - private final Configuration artemisConfiguration; + protected EmbeddedActiveMQ embeddedBrokerService; + private Configuration artemisConfiguration; private Consumer<Configuration> customConfigurator; private final int port; @@ -175,7 +175,15 @@ public abstract class AbstractArtemisEmbeddedService implements ArtemisInfraServ @Override public synchronized void initialize() { try { - if (embeddedBrokerService.getActiveMQServer() == null || !embeddedBrokerService.getActiveMQServer().isStarted()) { + ActiveMQServer existingServer = embeddedBrokerService.getActiveMQServer(); + if (existingServer == null || !existingServer.isStarted()) { + if (existingServer != null) { + // The service was already started and then stopped (e.g. when the tests are re-run after a failure). + // EmbeddedActiveMQ would reuse the stopped server with its previous configuration, which does not + // match the new broker id/address computed by configure(), so start from a clean state. + embeddedBrokerService = new EmbeddedActiveMQ(); + artemisConfiguration = new ConfigurationImpl(); + } embeddedBrokerService.setConfiguration(configure(port)); embeddedBrokerService.start();
