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 f240b9fa9ac104939d7dccf78e203ac6ccf71d44 Author: Aurélien Pupier <[email protected]> AuthorDate: Thu Oct 8 11:26:27 2026 +0200 CAMEL-25342 - Reduce flakiness of NATS ITs by waiting for consumers to be active before sending Core NATS does not persist messages for inactive subscribers: any message published before a dispatcher has finished subscribing is silently dropped. All integration tests that rely on a NATS consumer route receiving messages were subject to this race between the test body and the consumer's background executor thread. Add a waitForNatsConsumers(int count) helper to NatsITSupport (and delegate from NatsAuthITSupport / NatsAuthTokenITSupport which have an independent hierarchy) that blocks until at least the expected number of NatsConsumer instances report isActive(). Every test that sends to a core NATS topic now calls this helper before publishing, eliminating the race. Co-authored-by: IBM Bob 2.2.1 Signed-off-by: Aurélien Pupier <[email protected]> --- .../nats/integration/NatsAuthConsumerLoadIT.java | 2 ++ .../nats/integration/NatsAuthITSupport.java | 4 ++++ .../integration/NatsAuthTokenConsumerLoadIT.java | 2 ++ .../nats/integration/NatsAuthTokenITSupport.java | 3 +++ .../integration/NatsConsumerHeadersSupportIT.java | 2 ++ .../component/nats/integration/NatsConsumerIT.java | 13 +---------- .../nats/integration/NatsConsumerLoadIT.java | 2 ++ .../integration/NatsConsumerMaxMessagesIT.java | 3 +++ .../NatsConsumerMaxMessagesQueueIT.java | 2 ++ .../integration/NatsConsumerReplyToFailureIT.java | 6 ++++++ .../nats/integration/NatsConsumerReplyToIT.java | 2 ++ .../integration/NatsConsumerReplyToInOnlyIT.java | 2 ++ .../NatsConsumerWithConnectionLoadIT.java | 2 ++ .../integration/NatsConsumerWithRedeliveryIT.java | 11 +--------- .../component/nats/integration/NatsITSupport.java | 25 ++++++++++++++++++++++ .../nats/integration/NatsProducerReplyToIT.java | 2 ++ .../nats/jetstream/NatsJetstreamConsumerIT.java | 2 ++ 17 files changed, 63 insertions(+), 22 deletions(-) diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthConsumerLoadIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthConsumerLoadIT.java index 372137e5e1a0..b50362e131aa 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthConsumerLoadIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthConsumerLoadIT.java @@ -37,6 +37,8 @@ public class NatsAuthConsumerLoadIT extends NatsAuthITSupport { Options options = new Options.Builder().server("nats://" + service.getServiceAddress()).build(); Connection connection = Nats.connect(options); + waitForNatsConsumers(1); + for (int i = 0; i < 100; i++) { connection.publish("test", ("test" + i).getBytes()); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthITSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthITSupport.java index b4661f1e627e..6e5686c5f7e8 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthITSupport.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthITSupport.java @@ -34,4 +34,8 @@ public class NatsAuthITSupport extends CamelTestSupport { nats.getConfiguration().setServers(service.getServiceAddress()); return context; } + + protected void waitForNatsConsumers(int count) { + NatsITSupport.waitForNatsConsumers(context, count); + } } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenConsumerLoadIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenConsumerLoadIT.java index add81ec62434..77060e36c8fb 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenConsumerLoadIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenConsumerLoadIT.java @@ -37,6 +37,8 @@ public class NatsAuthTokenConsumerLoadIT extends NatsAuthTokenITSupport { Options options = new Options.Builder().server("nats://" + service.getServiceAddress()).build(); Connection connection = Nats.connect(options); + waitForNatsConsumers(1); + for (int i = 0; i < 100; i++) { connection.publish("test", ("test" + i).getBytes()); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenITSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenITSupport.java index 198a9377dd5c..045799193b04 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenITSupport.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsAuthTokenITSupport.java @@ -34,4 +34,7 @@ public class NatsAuthTokenITSupport extends CamelTestSupport { return context; } + protected void waitForNatsConsumers(int count) { + NatsITSupport.waitForNatsConsumers(context, count); + } } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerHeadersSupportIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerHeadersSupportIT.java index 654c18622039..25b1f85155e4 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerHeadersSupportIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerHeadersSupportIT.java @@ -54,6 +54,8 @@ class NatsConsumerHeadersSupportIT extends NatsITSupport { this.mockResultEndpoint.expectedHeaderReceived(HEADER_KEY_1, HEADER_VALUE_1); this.mockResultEndpoint.expectedHeaderReceived(HEADER_KEY_2, secondHeaders); + waitForNatsConsumers(1); + final Options options = new Options.Builder().server("nats://" + service.getServiceAddress()).build(); final Connection connection = Nats.connect(options); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerIT.java index 6b75bb47349c..2c53f6f39541 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerIT.java @@ -16,18 +16,12 @@ */ package org.apache.camel.component.nats.integration; -import java.util.concurrent.TimeUnit; - import org.apache.camel.EndpointInject; -import org.apache.camel.Route; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.component.nats.NatsConstants; -import org.apache.camel.component.nats.NatsConsumer; import org.junit.jupiter.api.Test; -import static org.awaitility.Awaitility.await; - public class NatsConsumerIT extends NatsITSupport { @EndpointInject("mock:result") @@ -38,12 +32,7 @@ public class NatsConsumerIT extends NatsITSupport { mockResultEndpoint.expectedBodiesReceived("Hello World"); mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT, "test"); - // Wait for the NATS consumer to be subscribed before sending messages, - // since core NATS does not persist messages for inactive subscribers - await().atMost(10, TimeUnit.SECONDS) - .until(() -> context.getRoutes().stream() - .map(Route::getConsumer) - .anyMatch(c -> c instanceof NatsConsumer nc && nc.isActive())); + waitForNatsConsumers(1); template.sendBody("direct:send", "Hello World"); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerLoadIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerLoadIT.java index ab88972b084f..e6006828a541 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerLoadIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerLoadIT.java @@ -35,6 +35,8 @@ public class NatsConsumerLoadIT extends NatsITSupport { Options options = new Options.Builder().server("nats://" + service.getServiceAddress()).build(); Connection connection = Nats.connect(options); + waitForNatsConsumers(1); + for (int i = 0; i < 10000; i++) { connection.publish("test", ("test" + i).getBytes()); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesIT.java index d480aa9613e1..145988ccb883 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesIT.java @@ -29,6 +29,9 @@ public class NatsConsumerMaxMessagesIT extends NatsITSupport { @Test public void testMaxConsumer() throws InterruptedException { mockResultEndpoint.setExpectedMessageCount(5); + + waitForNatsConsumers(1); + template.sendBody("direct:send", "test"); template.sendBody("direct:send", "test1"); template.sendBody("direct:send", "test2"); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesQueueIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesQueueIT.java index 65b3df4c4b3f..54fc2b581a5d 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesQueueIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerMaxMessagesQueueIT.java @@ -30,6 +30,8 @@ public class NatsConsumerMaxMessagesQueueIT extends NatsITSupport { public void testMaxConsumer() throws InterruptedException { mockResultEndpoint.setExpectedMessageCount(2); + waitForNatsConsumers(2); + template.sendBody("direct:send", "test"); template.sendBody("direct:send", "test1"); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToFailureIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToFailureIT.java index eb23a0fe84c7..0227d4bcaee9 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToFailureIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToFailureIT.java @@ -19,6 +19,7 @@ package org.apache.camel.component.nats.integration; import org.apache.camel.CamelExecutionException; import org.apache.camel.ExchangeTimedOutException; import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.apache.camel.test.junit6.TestSupport.assertIsInstanceOf; @@ -31,6 +32,11 @@ import static org.junit.jupiter.api.Assertions.assertThrows; */ public class NatsConsumerReplyToFailureIT extends NatsITSupport { + @BeforeEach + public void waitForConsumers() { + waitForNatsConsumers(2); + } + @Test public void testNoReplyWhenExchangeFailed() { CamelExecutionException e = assertThrows(CamelExecutionException.class, diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToIT.java index 7a640ba5a3b5..f1ae931dde13 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToIT.java @@ -37,6 +37,8 @@ public class NatsConsumerReplyToIT extends NatsITSupport { mockReplyEndpoint.expectedBodiesReceived("Bye World"); mockReplyEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT, "myReplyQueue"); + waitForNatsConsumers(2); + template.sendBody("direct:send", "World"); mockResultEndpoint.setAssertPeriod(5000); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToInOnlyIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToInOnlyIT.java index f259f66546be..845fd09f44b6 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToInOnlyIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerReplyToInOnlyIT.java @@ -35,6 +35,8 @@ public class NatsConsumerReplyToInOnlyIT extends NatsITSupport { // reply endpoint should NOT receive any message when exchange pattern is InOnly mockReplyEndpoint.expectedMessageCount(0); + waitForNatsConsumers(2); + template.sendBody("direct:send", "World"); mockResultEndpoint.setAssertPeriod(5000); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithConnectionLoadIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithConnectionLoadIT.java index 0d08ac4a7aab..c2e480d01458 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithConnectionLoadIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithConnectionLoadIT.java @@ -51,6 +51,8 @@ public class NatsConsumerWithConnectionLoadIT extends NatsITSupport { Options options = new Options.Builder().server("nats://" + service.getServiceAddress()).build(); Connection connection = Nats.connect(options); + waitForNatsConsumers(1); + for (int i = 0; i < 100; i++) { connection.publish("test", ("test" + i).getBytes()); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithRedeliveryIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithRedeliveryIT.java index 2325041eb91b..ed68bf26b321 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithRedeliveryIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsConsumerWithRedeliveryIT.java @@ -20,15 +20,11 @@ import java.util.concurrent.TimeUnit; import org.apache.camel.EndpointInject; import org.apache.camel.LoggingLevel; -import org.apache.camel.Route; import org.apache.camel.RuntimeCamelException; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; -import org.apache.camel.component.nats.NatsConsumer; import org.junit.jupiter.api.Test; -import static org.awaitility.Awaitility.await; - public class NatsConsumerWithRedeliveryIT extends NatsITSupport { private static final int REDELIVERY_COUNT = 2; @@ -44,12 +40,7 @@ public class NatsConsumerWithRedeliveryIT extends NatsITSupport { mockResultEndpoint.setExpectedMessageCount(1); exception.setExpectedMessageCount(1); - // Wait for the NATS consumer to be subscribed before sending messages, - // since core NATS does not persist messages for inactive subscribers - await().atMost(10, TimeUnit.SECONDS) - .until(() -> context.getRoutes().stream() - .map(Route::getConsumer) - .anyMatch(c -> c instanceof NatsConsumer nc && nc.isActive())); + waitForNatsConsumers(1); template.sendBody("direct:send", "test"); template.sendBody("direct:send", "golang"); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsITSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsITSupport.java index b3ba6c165d2e..7ccd437ad34e 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsITSupport.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsITSupport.java @@ -18,10 +18,13 @@ package org.apache.camel.component.nats.integration; import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.TimeUnit; import java.util.logging.LogManager; import org.apache.camel.CamelContext; +import org.apache.camel.Route; import org.apache.camel.component.nats.NatsComponent; +import org.apache.camel.component.nats.NatsConsumer; import org.apache.camel.test.infra.nats.services.NatsService; import org.apache.camel.test.infra.nats.services.NatsServiceFactory; import org.apache.camel.test.junit6.CamelTestSupport; @@ -29,6 +32,8 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class NatsITSupport extends CamelTestSupport { @RegisterExtension static NatsService service = NatsServiceFactory.createSingletonService(); @@ -52,4 +57,24 @@ public class NatsITSupport extends CamelTestSupport { nats.getConfiguration().setServers(service.getServiceAddress()); return context; } + + /** + * Waits until at least {@code count} NATS consumers in the current context have finished subscribing and are + * active. Core NATS does not persist messages for inactive subscribers, so callers must invoke this before + * publishing or sending any message that a consumer route is expected to receive. + */ + protected void waitForNatsConsumers(int count) { + waitForNatsConsumers(context, count); + } + + /** + * Static variant for use in support classes that do not extend {@link NatsITSupport}. + */ + static void waitForNatsConsumers(CamelContext ctx, int count) { + await().atMost(10, TimeUnit.SECONDS) + .until(() -> ctx.getRoutes().stream() + .map(Route::getConsumer) + .filter(c -> c instanceof NatsConsumer nc && nc.isActive()) + .count() >= count); + } } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsProducerReplyToIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsProducerReplyToIT.java index 7a012494dcf6..5b3b0ebb97ad 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsProducerReplyToIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/integration/NatsProducerReplyToIT.java @@ -37,6 +37,8 @@ public class NatsProducerReplyToIT extends NatsITSupport { resultEndpoint = context.getEndpoint(resultUri, MockEndpoint.class); resultEndpoint.expectedBodiesReceivedInAnyOrder("Bye Camel", "Bye World"); + waitForNatsConsumers(1); + String out = template.requestBody(startUri, body1, String.class); String out2 = template.requestBody(startUri, body2, String.class); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerIT.java index 84eb5cbd7bce..1ed351b6bc0e 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerIT.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerIT.java @@ -35,6 +35,8 @@ public class NatsJetstreamConsumerIT extends NatsITSupport { mockResultEndpoint.expectedBodiesReceived("Hello World"); mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT, "mytopic"); + waitForNatsConsumers(1); + template.sendBody("direct:send", "Hello World"); mockResultEndpoint.setAssertPeriod(5000);
