This is an automated email from the ASF dual-hosted git repository.
gnodet 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 1aaaa3766225 CAMEL-23889: Fix flaky camel-vertx-websocket surefire
tests
1aaaa3766225 is described below
commit 1aaaa3766225b3d3cfcb654084258fbeba9f5a74
Author: Guillaume Nodet <[email protected]>
AuthorDate: Sat Jul 4 12:12:57 2026 +0200
CAMEL-23889: Fix flaky camel-vertx-websocket surefire tests
* Fix flaky camel-vertx-websocket surefire tests
Replace Thread.sleep() calls with Awaitility-based async assertions
in the vertx-websocket reconnect tests, and fix an incorrect timeout
value in VertxWebSocketEventTest.
Changes:
- VertxWebsocketConsumerAsClientReconnectTest: Replace Thread.sleep(300)
with Awaitility await().untilAsserted() using fresh WebSocket
connections to bypass stale producer endpoint cache after server
restart.
- VertxWebsocketConsumerAsClientMaxReconnectTest: Replace
Thread.sleep(300) with Awaitility during(2, SECONDS) pattern to
reliably verify reconnect attempts are exhausted.
- VertxWebSocketEventTest: Fix timeout from 5000 seconds (83 minutes)
to 5 seconds in webSocketFuture.get() call.
- Add Awaitility test dependency to camel-vertx-websocket pom.xml.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
* Simplify MaxReconnect test: replace during() with pollDelay()
The during().untilAsserted() pattern was correct but misleading — with
expectedMessageCount(0), assertIsSatisfied() passes immediately on every
poll, so during() was effectively just a timed wait. Use pollDelay()
instead which more clearly expresses the intent: wait for reconnect
attempts to exhaust, then verify.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
* Fix pre-existing race in "verify send fails" assertions
After stopRoute("server"), the producer endpoint's cached WebSocket may
still appear open (isClosed() returns false) until the Vert.x event loop
processes the TCP close frame. Wrap the "verify that we cannot send
messages" assertion in Awaitility to wait until the close propagates,
and restore full end-to-end Camel flow in the reconnect test.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
* CAMEL-23889: Address review feedback on MockEndpoint assertions
- ReconnectTest: assertIsSatisfied(500) timeout was silently ignored
when expectedCount > 0 (defaults to 10s latch wait). Use
setResultWaitTime(500) to control the actual wait time.
- MaxReconnectTest: Replace Awaitility pollDelay wrapper with direct
assertIsSatisfied(2000) for the zero-count wait (simpler, equivalent).
Use setAssertPeriod(1000) for final assertion to re-verify after a
delay and catch late deliveries.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---------
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../camel-vertx/camel-vertx-websocket/pom.xml | 6 +++
.../vertx/websocket/VertxWebSocketEventTest.java | 2 +-
...xWebsocketConsumerAsClientMaxReconnectTest.java | 38 ++++++++++------
...ertxWebsocketConsumerAsClientReconnectTest.java | 50 +++++++++++++++-------
4 files changed, 67 insertions(+), 29 deletions(-)
diff --git a/components/camel-vertx/camel-vertx-websocket/pom.xml
b/components/camel-vertx/camel-vertx-websocket/pom.xml
index 109d111806c9..b7590aab6f32 100644
--- a/components/camel-vertx/camel-vertx-websocket/pom.xml
+++ b/components/camel-vertx/camel-vertx-websocket/pom.xml
@@ -55,6 +55,12 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-main</artifactId>
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebSocketEventTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebSocketEventTest.java
index a998d451e0bd..d4cfa4969061 100644
---
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebSocketEventTest.java
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebSocketEventTest.java
@@ -52,7 +52,7 @@ public class VertxWebSocketEventTest extends
VertxWebSocketTestSupport {
template.sendBody("vertx-websocket:localhost:" + port + "/test",
MESSAGE_BODY);
- ServerWebSocket webSocket = webSocketFuture.get(5000,
TimeUnit.SECONDS);
+ ServerWebSocket webSocket = webSocketFuture.get(5, TimeUnit.SECONDS);
assertNotNull(webSocket);
// Trigger error event (message length > max allowed)
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientMaxReconnectTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientMaxReconnectTest.java
index 1d5a173eb7d2..bdf1ecf4ed48 100644
---
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientMaxReconnectTest.java
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientMaxReconnectTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.vertx.websocket;
import java.net.ConnectException;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -26,6 +27,8 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
+
public class VertxWebsocketConsumerAsClientMaxReconnectTest extends
VertxWebSocketTestSupport {
@Test
void testMaxReconnect() throws Exception {
@@ -42,25 +45,36 @@ public class VertxWebsocketConsumerAsClientMaxReconnectTest
extends VertxWebSock
context.getRouteController().stopRoute("server");
- // Verify that we cannot send messages
- Exchange exchange = template.send(uri, new Processor() {
- @Override
- public void process(Exchange exchange) throws Exception {
- exchange.getMessage().setBody("Hello World Again");
- }
- });
- Exception exception = exchange.getException();
- Assertions.assertNotNull(exception);
- Assertions.assertInstanceOf(ConnectException.class,
exception.getCause());
+ // Verify that the server is fully down by waiting until sends fail.
+ // The producer endpoint's cached WebSocket may still appear open
briefly
+ // after stopRoute returns, until the Vert.x event loop processes the
+ // TCP close frame and isClosed() starts returning true.
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ Exchange exchange = template.send(uri, new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ exchange.getMessage().setBody("Hello World Again");
+ }
+ });
+ Assertions.assertNotNull(exchange.getException());
+ Assertions.assertInstanceOf(ConnectException.class,
exchange.getException().getCause());
+ });
// Wait for client consumer reconnect max attempts to be exhausted
- Thread.sleep(300);
+ // (maxReconnectAttempts=1, reconnectInterval=10ms).
+ // With expectedMessageCount(0), assertIsSatisfied(2000) sleeps for
+ // 2 seconds then verifies no messages were received.
+ mockEndpoint.assertIsSatisfied(2000);
// Restart server
context.getRouteController().startRoute("server");
- // Verify that the client consumer gave up reconnecting
+ // Verify that the client consumer gave up reconnecting.
+ // Use setAssertPeriod to re-verify after a delay, catching any
+ // late deliveries that arrive after the initial assertion passes.
template.sendBody(uri, "Hello World Again");
+ mockEndpoint.setAssertPeriod(1000);
mockEndpoint.assertIsSatisfied();
}
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientReconnectTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientReconnectTest.java
index 1380be718dc7..e4d2de3e5911 100644
---
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientReconnectTest.java
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientReconnectTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.vertx.websocket;
import java.net.ConnectException;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -26,6 +27,8 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
+
public class VertxWebsocketConsumerAsClientReconnectTest extends
VertxWebSocketTestSupport {
@Test
void testReconnect() throws Exception {
@@ -42,26 +45,41 @@ public class VertxWebsocketConsumerAsClientReconnectTest
extends VertxWebSocketT
context.getRouteController().stopRoute("server");
- // Verify that we cannot send messages
- Exchange exchange = template.send(uri, new Processor() {
- @Override
- public void process(Exchange exchange) throws Exception {
- exchange.getMessage().setBody("Hello World Again");
- }
- });
- Exception exception = exchange.getException();
- Assertions.assertNotNull(exception);
- Assertions.assertInstanceOf(ConnectException.class,
exception.getCause());
+ // Verify that the server is fully down by waiting until sends fail.
+ // The producer endpoint's cached WebSocket may still appear open
briefly
+ // after stopRoute returns, until the Vert.x event loop processes the
+ // TCP close frame and isClosed() starts returning true.
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ Exchange exchange = template.send(uri, new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ exchange.getMessage().setBody("Hello World Again");
+ }
+ });
+ Assertions.assertNotNull(exchange.getException());
+ Assertions.assertInstanceOf(ConnectException.class,
exchange.getException().getCause());
+ });
// Restart server
context.getRouteController().startRoute("server");
- // Wait for client consumer reconnect
- Thread.sleep(300);
-
- // Verify that the client consumer reconnected
- template.sendBody(uri, "Hello World Again");
- mockEndpoint.assertIsSatisfied();
+ // Wait for client consumer to reconnect and verify end-to-end message
flow.
+ // After the server stops, the client consumer's close handler fires
+ // asynchronously on the Vert.x event loop, starting the reconnect
timer
+ // that connects to the restarted server.
+ // Use ignoreExceptions() to retry through transient failures while
both
+ // the producer and client consumer re-establish their connections.
+ await().atMost(20, TimeUnit.SECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .ignoreExceptions()
+ .untilAsserted(() -> {
+ mockEndpoint.reset();
+ mockEndpoint.expectedBodiesReceived("Hello World Again");
+ mockEndpoint.setResultWaitTime(500);
+ template.sendBody(uri, "Hello World Again");
+ mockEndpoint.assertIsSatisfied();
+ });
}
@Override
