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
The following commit(s) were added to refs/heads/main by this push:
new 86304c0f4d8d CAMEL-26366: camel-nats - Fix flaky push consumer tests
introduced by CAMEL-25845 and CAMEL-25847
86304c0f4d8d is described below
commit 86304c0f4d8d801cc4fdfa14bc06262e6b5ffabe
Author: Aurélien Pupier <[email protected]>
AuthorDate: Wed Oct 28 11:18:37 2026 +0100
CAMEL-26366: camel-nats - Fix flaky push consumer tests introduced by
CAMEL-25845 and CAMEL-25847
Two tests are flaky on CI:
- NatsConsumerPushConsumerDeletedTest
.removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes
("no keeper listens to the connection after the stop ==> expected: <[]>
but was: <[NatsJetStreamPushConsumerKeeper@...]>")
- NatsConsumerJetStreamPushStopTest
.stopWhileSubscribingIsNotReportedAsAnError
("no subscription left once the consumer stopped ==> expected: <0> but
was: <1>")
Cause: the consumer submits its first subscribe on its own executor in
doStart(), so it runs asynchronously. The tests started to change the
state of the consumer (replace the route, install the beforeSubscribe
hook, stop and start the consumer) without waiting for that first
subscribe to complete. The first subscribe could then overlap with the
steps of the test, or consume the one-shot hook meant for the next
subscribe, and leave behind a connection listener or an active
subscription that nothing ends.
Fix (test only, no production change), with an Awaitility wait that has
an explicit atMost until the first subscribe is done:
- NatsConsumerPushConsumerDeletedTest waits for the connection listener
of the first subscribe at the start of the failing test.
- NatsConsumerJetStreamPushStopTest waits for the active subscription in
a @BeforeEach, so that every test of the class waits, as suggested in
the review: the class uses DeliverPolicy.New, so the other tests that
publish straight away have the same race (a message published before
the first subscribe is skipped).
Verification, running each of the two failing test methods 10 times in a
row (mvn test -Dtest=...) on the local machine:
- without the fix: removesItsConnectionListener... failed 7 times out of
10 (runs 2, 3, 5, 6, 7, 9 and 10, with the exact message of the JIRA;
most were passed by the surefire rerun and reported as flakes).
stopWhileSubscribingIsNotReportedAsAnError did not fail in 10 runs, so
it could not be reproduced locally: the wait added for it follows from
the analysis of the code and from the identical pattern of the other
test, and was not confirmed by a local failure. It is likely to need
a slower or loaded machine like CI to show up.
- with the fix: 0 failures out of 10 runs for both tests, and 0 failures
out of 5 runs of the two whole test classes (17 tests each run).
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../camel/component/nats/NatsConsumerJetStreamPushStopTest.java | 9 +++++++++
.../component/nats/NatsConsumerPushConsumerDeletedTest.java | 3 +++
2 files changed, 12 insertions(+)
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
index ea1f942f82b4..10f7f123e8ff 100644
---
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
@@ -52,6 +52,7 @@ import org.apache.camel.spi.ExceptionHandler;
import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.awaitility.Awaitility.await;
@@ -73,6 +74,14 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
private final AtomicBoolean slowInterrupted = new AtomicBoolean();
private final AtomicBoolean slowDone = new AtomicBoolean();
+ @BeforeEach
+ void waitForTheFirstSubscribe() {
+ // the route subscribes on the executor of its consumer, and the
consumer uses DeliverPolicy.New: a message
+ // published before that subscribe is skipped, and a stop or a start
before it overlaps with it (it would
+ // leave a subscription that nobody ends)
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
server.activeSubscriptions() == 1);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
context.getRegistry().bind("conn", server.connection());
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
index f663f04fe437..fa70205f05d1 100644
---
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
@@ -212,6 +212,9 @@ class NatsConsumerPushConsumerDeletedTest extends
CamelTestSupport {
@Test
void removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes()
throws Exception {
+ // the route of the test subscribes on the executor of its consumer:
wait for it, so that the hook below
+ // applies to the subscribe of the route that replaces it, not to this
first one
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!server.listeners.isEmpty());
int added = server.listenersAdded.get();
// the route is stopped while the consumer makes its first
subscription, so the stop does not see the keeper
// yet: the keeper must not stay registered on the (shared) connection
