This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 b0fb55a79e20 CAMEL-23918: Fix flaky tests in camel-reactive-streams
b0fb55a79e20 is described below
commit b0fb55a79e2002ba532878cc943b46b5a5cca0f8
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Jul 8 07:09:23 2026 +0200
CAMEL-23918: Fix flaky tests in camel-reactive-streams
Replace Thread.sleep calls with Awaitility and Camel-idiomatic patterns
across 5 test classes. Fix a race condition in BackpressureStrategyTest
where the async flush in CamelSubscription caused exact-value assertions
to fail non-deterministically — replaced with invariant-based assertions
that verify actual backpressure strategy behavior (OLDEST keeps first
item, LATEST keeps most recent).
Closes #24448
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
components/camel-reactive-streams/pom.xml | 6 ++
.../BackpressurePublisherRoutePolicyTest.java | 13 ++--
.../reactive/streams/BackpressureStrategyTest.java | 79 +++++++++++++++++-----
.../streams/BackpressureSubscriberTest.java | 6 +-
.../reactive/streams/BasicPublisherTest.java | 5 +-
.../reactive/streams/DelayedMonoPublisherTest.java | 36 +++++-----
6 files changed, 97 insertions(+), 48 deletions(-)
diff --git a/components/camel-reactive-streams/pom.xml
b/components/camel-reactive-streams/pom.xml
index 73ad33c8fa20..348873ed66d0 100644
--- a/components/camel-reactive-streams/pom.xml
+++ b/components/camel-reactive-streams/pom.xml
@@ -87,6 +87,12 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
index c3c603012c73..b61e678d1ac0 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
@@ -30,6 +30,7 @@ import
org.apache.camel.throttling.ThrottlingInflightRoutePolicy;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -82,11 +83,11 @@ public class BackpressurePublisherRoutePolicyTest extends
BaseReactiveTest {
// fire a delayed request from the subscriber (required by camel core)
subscriber.request(1);
- Thread.sleep(250);
+ // Wait for the route policy to stop or suspend the route
StatefulService service = (StatefulService)
context().getRoute("policy-route").getConsumer();
- // ensure the route is stopped or suspended
- assertTrue(service.isStopped() || service.isSuspended());
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertTrue(service.isStopped() ||
service.isSuspended()));
// request all the remaining exchanges
subscriber.request(24);
@@ -141,11 +142,11 @@ public class BackpressurePublisherRoutePolicyTest extends
BaseReactiveTest {
// fire a delayed request from the subscriber (required by camel core)
subscriber.request(1);
- Thread.sleep(250);
+ // Wait for the route policy to stop or suspend the route
StatefulService service = (StatefulService)
context().getRoute("policy-route").getConsumer();
- // ensure the route is stopped or suspended
- assertTrue(service.isStopped() || service.isSuspended());
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertTrue(service.isStopped() ||
service.isSuspended()));
subscriber.cancel();
// request other exchanges to ensure that the route works
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
index 46c52aa23262..53c9ba7ac865 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
@@ -16,9 +16,13 @@
*/
package org.apache.camel.component.reactive.streams;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import io.reactivex.Flowable;
import org.apache.camel.Exchange;
@@ -27,6 +31,7 @@ import
org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
import org.apache.camel.component.reactive.streams.support.TestSubscriber;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -73,11 +78,14 @@ public class BackpressureStrategyTest extends
BaseReactiveTest {
ReactiveStreamsComponent comp = (ReactiveStreamsComponent)
context().getComponent("reactive-streams");
comp.setBackpressureStrategy(ReactiveStreamsBackpressureStrategy.OLDEST);
+ AtomicInteger timerCount = new AtomicInteger(0);
+
new RouteBuilder() {
@Override
public void configure() {
from("timer:gen?period=20&repeatCount=20&includeMetadata=true")
.setBody().header(Exchange.TIMER_COUNTER)
+ .process(e -> timerCount.incrementAndGet())
.to("reactive-streams:integers");
}
}.addRoutesToCamelContext(context);
@@ -100,14 +108,25 @@ public class BackpressureStrategyTest extends
BaseReactiveTest {
context().start();
assertTrue(latch.await(5, TimeUnit.SECONDS));
- Thread.sleep(1000); // wait for all numbers to be generated
+ // Wait for all 20 timer events to be processed
+ await().atMost(5, TimeUnit.SECONDS).until(() -> timerCount.get() >=
20);
subscriber.request(19);
- assertTrue(latch2.await(1, TimeUnit.SECONDS));
- Thread.sleep(200); // add other time to ensure no other items arrive
- assertEquals(2, queue.size());
- int sum = queue.stream().reduce((i, j) -> i + j).get();
- assertEquals(3, sum); // 1 + 2 = 3
+ assertTrue(latch2.await(2, TimeUnit.SECONDS));
+ // Verify exactly 2 items received and no more arrive
+ // Use during() to ensure queue size remains stable at 2
+ await().during(Duration.ofMillis(500))
+ .atMost(Duration.ofSeconds(2))
+ .until(() -> queue.size() == 2);
+ // Verify OLDEST strategy behavior:
+ // - First item is always 1 (OLDEST keeps the first buffered item)
+ // - Second item is an early value (the first item buffered after the
initial flush)
+ // The exact second value depends on flush timing vs timer events,
+ // but it must be > 1 and much less than 20 (the last timer event)
+ List<Integer> items = new ArrayList<>(queue);
+ assertEquals(1, items.get(0).intValue(), "OLDEST strategy should
deliver item 1 first");
+ assertTrue(items.get(1) > 1 && items.get(1) < 20,
+ "OLDEST strategy should keep an early buffered item, got: " +
items.get(1));
subscriber.cancel();
}
@@ -118,11 +137,14 @@ public class BackpressureStrategyTest extends
BaseReactiveTest {
ReactiveStreamsComponent comp = (ReactiveStreamsComponent)
context().getComponent("reactive-streams");
comp.setBackpressureStrategy(ReactiveStreamsBackpressureStrategy.LATEST);
+ AtomicInteger timerCount = new AtomicInteger(0);
+
new RouteBuilder() {
@Override
public void configure() {
from("timer:gen?period=20&repeatCount=20&includeMetadata=true")
.setBody().header(Exchange.TIMER_COUNTER)
+ .process(e -> timerCount.incrementAndGet())
.to("reactive-streams:integers");
}
}.addRoutesToCamelContext(context);
@@ -145,25 +167,38 @@ public class BackpressureStrategyTest extends
BaseReactiveTest {
context().start();
assertTrue(latch.await(5, TimeUnit.SECONDS));
- Thread.sleep(1000); // wait for all numbers to be generated
+ // Wait for all 20 timer events to be processed
+ await().atMost(5, TimeUnit.SECONDS).until(() -> timerCount.get() >=
20);
subscriber.request(19);
- assertTrue(latch2.await(1, TimeUnit.SECONDS));
- Thread.sleep(200); // add other time to ensure no other items arrive
- assertEquals(2, queue.size());
- int sum = queue.stream().reduce((i, j) -> i + j).get();
- assertEquals(21, sum); // 1 + 20 = 21
+ assertTrue(latch2.await(2, TimeUnit.SECONDS));
+ // Verify exactly 2 items received and no more arrive
+ // Use during() to ensure queue size remains stable at 2
+ await().during(Duration.ofMillis(500))
+ .atMost(Duration.ofSeconds(2))
+ .until(() -> queue.size() == 2);
+ // Verify LATEST strategy behavior:
+ // - Second item is always 20 (LATEST always replaces with the newest)
+ // - First item depends on flush timing but is always < 20
+ List<Integer> items = new ArrayList<>(queue);
+ assertTrue(items.get(0) >= 1 && items.get(0) < 20,
+ "First item should be less than 20, got: " + items.get(0));
+ assertEquals(20, items.get(1).intValue(), "LATEST strategy should keep
the most recent item (20)");
subscriber.cancel();
}
@Test
public void testBackpressureDropStrategyInEndpoint() throws Exception {
+
+ AtomicInteger timerCount = new AtomicInteger(0);
+
new RouteBuilder() {
@Override
public void configure() {
from("timer:gen?period=20&repeatCount=20&includeMetadata=true")
.setBody().header(Exchange.TIMER_COUNTER)
+ .process(e -> timerCount.incrementAndGet())
.to("reactive-streams:integers?backpressureStrategy=OLDEST");
}
}.addRoutesToCamelContext(context);
@@ -186,14 +221,22 @@ public class BackpressureStrategyTest extends
BaseReactiveTest {
context().start();
assertTrue(latch.await(5, TimeUnit.SECONDS));
- Thread.sleep(1000); // wait for all numbers to be generated
+ // Wait for all 20 timer events to be processed
+ await().atMost(5, TimeUnit.SECONDS).until(() -> timerCount.get() >=
20);
subscriber.request(19);
- assertTrue(latch2.await(1, TimeUnit.SECONDS));
- Thread.sleep(200); // add other time to ensure no other items arrive
- assertEquals(2, queue.size());
- int sum = queue.stream().reduce((i, j) -> i + j).get();
- assertEquals(3, sum); // 1 + 2 = 3
+ assertTrue(latch2.await(2, TimeUnit.SECONDS));
+ // Verify exactly 2 items received and no more arrive
+ // Use during() to ensure queue size remains stable at 2
+ await().during(Duration.ofMillis(500))
+ .atMost(Duration.ofSeconds(2))
+ .until(() -> queue.size() == 2);
+ // Verify OLDEST strategy behavior (same invariant as
testBackpressureDropStrategy):
+ // - First item is always 1, second is an early buffered item
+ List<Integer> items = new ArrayList<>(queue);
+ assertEquals(1, items.get(0).intValue(), "OLDEST strategy should
deliver item 1 first");
+ assertTrue(items.get(1) > 1 && items.get(1) < 20,
+ "OLDEST strategy should keep an early buffered item, got: " +
items.get(1));
subscriber.cancel();
}
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
index 591138817804..70eb55b3aa95 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
@@ -93,15 +93,15 @@ public class BackpressureSubscriberTest extends
BaseReactiveTest {
@Override
public void configure() {
from("reactive-streams:slowNumbers?concurrentConsumers=10&maxInflightExchanges=1")
- .process(x -> Thread.sleep(50))
+ .delay(50)
.to("mock:endpoint");
from("reactive-streams:slowerNumbers?concurrentConsumers=10&maxInflightExchanges=1")
- .process(x -> Thread.sleep(300))
+ .delay(300)
.to("mock:endpoint");
from("reactive-streams:parallelSlowNumbers?concurrentConsumers=10&maxInflightExchanges=5")
- .process(x -> Thread.sleep(100))
+ .delay(100)
.to("mock:endpoint");
}
};
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
index 594d8792d654..fc126168ee0b 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
@@ -95,10 +95,7 @@ public class BasicPublisherTest extends BaseReactiveTest {
disp1.dispose();
disp2.dispose();
- // No active subscriptions, warnings expected
- Thread.sleep(60);
-
- // Add another subscription
+ // Add another subscription after disposal
CountDownLatch latch3 = new CountDownLatch(5);
Disposable disp3 =
Observable.fromPublisher(CamelReactiveStreams.get(context).fromStream("unbounded",
Integer.class))
.subscribe(n -> latch3.countDown());
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DelayedMonoPublisherTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DelayedMonoPublisherTest.java
index f34ee0e7e40c..603ff9bf3ccf 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DelayedMonoPublisherTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DelayedMonoPublisherTest.java
@@ -20,8 +20,8 @@ import java.util.LinkedList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -33,13 +33,14 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class DelayedMonoPublisherTest {
- private ExecutorService service;
+ private ScheduledExecutorService service;
@BeforeEach
public void init() {
@@ -131,10 +132,9 @@ public class DelayedMonoPublisherTest {
.doOnComplete(latch::countDown)
.subscribe();
- Thread.sleep(200);
- pub.setData(5);
+ service.schedule(() -> pub.setData(5), 200, TimeUnit.MILLISECONDS);
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
assertEquals(1, data.size());
assertEquals(5, data.get(0).intValue());
@@ -158,10 +158,9 @@ public class DelayedMonoPublisherTest {
.doOnComplete(latch::countDown)
.subscribe();
- Thread.sleep(200);
- pub.setData(5);
+ service.schedule(() -> pub.setData(5), 200, TimeUnit.MILLISECONDS);
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
assertEquals(2, data.size());
for (Integer n : data) {
@@ -182,15 +181,17 @@ public class DelayedMonoPublisherTest {
.doOnComplete(latch::countDown)
.subscribe();
- Thread.sleep(200);
- pub.setData(5);
+ service.schedule(() -> pub.setData(5), 200, TimeUnit.MILLISECONDS);
+
+ // Wait for first subscriber to receive data before subscribing second
+ await().atMost(2, TimeUnit.SECONDS).until(() -> data.size() >= 1);
Flowable.fromPublisher(pub)
.doOnNext(data::add)
.doOnComplete(latch::countDown)
.subscribe();
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
assertEquals(2, data.size());
for (Integer n : data) {
@@ -215,8 +216,10 @@ public class DelayedMonoPublisherTest {
latch.countDown();
});
- Thread.sleep(200);
- pub.setException(ex);
+ service.schedule(() -> pub.setException(ex), 200,
TimeUnit.MILLISECONDS);
+
+ // Wait for first subscriber to receive exception before subscribing
second
+ await().atMost(2, TimeUnit.SECONDS).until(() -> exceptions.size() >=
1);
Flowable.fromPublisher(pub)
.subscribe(item -> {
@@ -225,7 +228,7 @@ public class DelayedMonoPublisherTest {
latch.countDown();
});
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
assertEquals(2, exceptions.size());
for (Throwable t : exceptions) {
@@ -251,10 +254,9 @@ public class DelayedMonoPublisherTest {
pub.subscribe(sub);
- Thread.sleep(100);
- sub.request(1);
+ service.schedule(() -> sub.request(1), 100, TimeUnit.MILLISECONDS);
- Integer res = queue.poll(1, TimeUnit.SECONDS);
+ Integer res = queue.poll(2, TimeUnit.SECONDS);
assertEquals(Integer.valueOf(2), res);
}
