From be8aa3d21e234a5d52487dc13d515f4d21cf541e Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 16 Jul 2026 14:21:41 +0000 Subject: [PATCH 1/2] CAMEL-24063: Fix flaky DirectProducerBlockingTest.testProducerBlocksResumeTest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test suspends a route then sends with block=true&timeout=2000. A background thread waits for the main thread to reach TIMED_WAITING, then resumes the route. Under CI load, detecting the thread state and resuming can take more than 2 s, causing: DirectConsumerNotAvailableException: No consumers available on endpoint: direct://suspended?block=true&timeout=2000 Fix: increase both the background thread's Awaitility timeout (2s→10s) and the sendBody block timeout (2000→10000ms) so there is enough headroom even on loaded CI nodes. Develocity evidence: testProducerBlocksResumeTest has 8 flaky runs out of 716 total on main (1.1% flaky rate). Co-Authored-By: Claude Opus 4.6 --- .../component/direct/DirectProducerBlockingTest.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java index 4ad06aa15819c..160777ee4ecb3 100644 --- a/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java @@ -85,8 +85,10 @@ public void testProducerBlocksResumeTest() throws Exception { public void run() { try { // Wait for the main thread to enter TIMED_WAITING state - // (blocked on condition in DirectComponent.getConsumer) - await().atMost(2, TimeUnit.SECONDS) + // (blocked on condition in DirectComponent.getConsumer). + // Use a generous timeout — on slow CI the thread state + // detection can take longer than 2 s. + await().atMost(10, TimeUnit.SECONDS) .pollInterval(10, TimeUnit.MILLISECONDS) .until(() -> mainThread.getState() == Thread.State.TIMED_WAITING); @@ -98,8 +100,10 @@ public void run() { } }); - // This call will block until the route is resumed by the background thread - template.sendBody("direct:suspended?block=true&timeout=2000", "hello world"); + // This call will block until the route is resumed by the background thread. + // Use a generous timeout so the background thread has enough headroom to + // detect the TIMED_WAITING state and resume the route even under CI load. + template.sendBody("direct:suspended?block=true&timeout=10000", "hello world"); assertMockEndpointsSatisfied(); From 7c2217ae21cd6e51684e9def8ae5f76289f5d5b0 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Tue, 21 Jul 2026 14:18:39 +0000 Subject: [PATCH 2/2] CAMEL-24063: Revert Awaitility-based DirectProducerBlockingTest to race-tolerant design MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Awaitility-based TIMED_WAITING approach (introduced in CAMEL-19549, Jun 25/29) imposed a strict thread-ordering dependency that introduced flakiness where the original code had none (stable for 6 years). Replace with ScheduledExecutorService.schedule() — same 200ms delay semantics as the original Thread.sleep(200), without using Thread.sleep: - If resume fires before sendBody starts blocking: route is already resumed, sendBody finds the consumer immediately (pass) - If resume fires while sendBody is blocking: sendBody gets unblocked (pass) Both race outcomes produce a passing test. Restore timeout=1000 (800ms of headroom is plenty). Co-Authored-By: Claude Opus 4.6 --- .../direct/DirectProducerBlockingTest.java | 45 +++++++------------ 1 file changed, 17 insertions(+), 28 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java index 160777ee4ecb3..a692a4685b416 100644 --- a/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerBlockingTest.java @@ -16,8 +16,8 @@ */ package org.apache.camel.component.direct; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.apache.camel.CamelExchangeException; @@ -28,7 +28,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -74,36 +73,26 @@ public void testProducerBlocksWithNoConsumers() throws Exception { @Test public void testProducerBlocksResumeTest() throws Exception { - getMockEndpoint("mock:result").expectedMessageCount(1); - context.getRouteController().suspendRoute("foo"); - Thread mainThread = Thread.currentThread(); - ExecutorService executor = Executors.newSingleThreadExecutor(); - executor.submit(new Runnable() { - @Override - public void run() { - try { - // Wait for the main thread to enter TIMED_WAITING state - // (blocked on condition in DirectComponent.getConsumer). - // Use a generous timeout — on slow CI the thread state - // detection can take longer than 2 s. - await().atMost(10, TimeUnit.SECONDS) - .pollInterval(10, TimeUnit.MILLISECONDS) - .until(() -> mainThread.getState() == Thread.State.TIMED_WAITING); - - log.info("Resuming consumer"); - context.getRouteController().resumeRoute("foo"); - } catch (Exception e) { - log.error("Error in background thread", e); - } + // Schedule route resume after 200ms. This is race-tolerant by design: + // - If resume fires before sendBody starts blocking: route is already + // resumed, sendBody finds the consumer immediately and succeeds + // - If resume fires while sendBody is blocking: sendBody gets unblocked + // Either outcome produces a passing test. + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + executor.schedule(() -> { + try { + log.info("Resuming consumer"); + context.getRouteController().resumeRoute("foo"); + } catch (Exception e) { + log.error("Error resuming route", e); } - }); + }, 200, TimeUnit.MILLISECONDS); + + getMockEndpoint("mock:result").expectedMessageCount(1); - // This call will block until the route is resumed by the background thread. - // Use a generous timeout so the background thread has enough headroom to - // detect the TIMED_WAITING state and resume the route even under CI load. - template.sendBody("direct:suspended?block=true&timeout=10000", "hello world"); + template.sendBody("direct:suspended?block=true&timeout=1000", "hello world"); assertMockEndpointsSatisfied();