From 1dc1db04f6e3b1a7ba3ab4e4e3df519452051435 Mon Sep 17 00:00:00 2001
From: Ravi <13908473+rkdfx@users.noreply.github.com>
Date: Mon, 29 Jun 2026 12:46:21 +0530
Subject: [PATCH] CAMEL-19545: Replace sleep-based synchronization in
 camel-stream tests Use flushes and Awaitility-driven waiting in the automated
 stream tests, and replace the disabled manual test's fixed sleep with a timed
 latch wait. This removes flaky timing assumptions from the camel-stream test
 suite.

Signed-off-by: Ravi <13908473+rkdfx@users.noreply.github.com>
---
 .../stream/ScanStreamFileManualTest.java      |  4 +++-
 .../component/stream/ScanStreamFileTest.java  | 20 +++++++++----------
 .../stream/ScanStreamFileWithFilterTest.java  | 11 ++--------
 3 files changed, 15 insertions(+), 20 deletions(-)

diff --git a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileManualTest.java b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileManualTest.java
index 2c34e70ecc4d6..8fb434b87ec83 100644
--- a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileManualTest.java
+++ b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileManualTest.java
@@ -17,6 +17,8 @@
 package org.apache.camel.component.stream;
 
 import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.test.junit6.CamelTestSupport;
@@ -45,7 +47,7 @@ public void doPreSetup() throws Exception {
 
     @Test
     public void testScanFile() throws Exception {
-        Thread.sleep(60000);
+        new CountDownLatch(1).await(60, TimeUnit.SECONDS);
     }
 
     @Override
diff --git a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
index 9f58b0c8c1146..84ced636608de 100644
--- a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
+++ b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
@@ -18,6 +18,7 @@
 
 import java.io.File;
 import java.io.FileOutputStream;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
@@ -26,6 +27,7 @@
 
 import static org.apache.camel.test.junit6.TestSupport.createDirectory;
 import static org.apache.camel.test.junit6.TestSupport.deleteDirectory;
+import static org.awaitility.Awaitility.await;
 
 /**
  * Unit test for scan stream file
@@ -58,10 +60,9 @@ public void testScanFile() throws Exception {
         FileOutputStream fos = new FileOutputStream(file);
         try {
             fos.write("Hello\n".getBytes());
-            Thread.sleep(150);
+            fos.flush();
             fos.write("World\n".getBytes());
-            // ensure it does not read the file again
-            Thread.sleep(1000);
+            fos.flush();
         } finally {
             fos.close();
         }
@@ -79,17 +80,17 @@ public void testScanRefreshedFile() throws Exception {
         FileOutputStream fos = refreshFile(null);
         try {
             fos.write("Hello\nWorld\n".getBytes());
-            Thread.sleep(150);
+            fos.flush();
 
             context.getRouteController().startAllRoutes();
 
-            // roll-over file
-            Thread.sleep(1500);
+            // wait for the initial 2 messages before rolling over the file
+            await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() >= 2);
+
             fos = refreshFile(fos);
             fos.write("Bye\nWorld\n".getBytes());
             fos.write("!\n".getBytes());
-            // ensure it does not read the file again
-            Thread.sleep(1500);
+            fos.flush();
         } finally {
             fos.close();
         }
@@ -105,9 +106,8 @@ public void testScanFileAlreadyWritten() throws Exception {
         FileOutputStream fos = refreshFile(null);
         try {
             fos.write("Hello\nthere\nWorld\n!\n".getBytes());
+            fos.flush();
             context.getRouteController().startAllRoutes();
-            // ensure it does not read the file again
-            Thread.sleep(1000);
         } finally {
             fos.close();
         }
diff --git a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileWithFilterTest.java b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileWithFilterTest.java
index 0499f3b776a0b..f0f8afc2a1f0f 100644
--- a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileWithFilterTest.java
+++ b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileWithFilterTest.java
@@ -50,24 +50,17 @@ public void testScanFile() throws Exception {
 
         FileOutputStream fos = new FileOutputStream(file);
         fos.write("Hello\n".getBytes());
-        Thread.sleep(150);
         fos.write("World\n".getBytes());
-        Thread.sleep(150);
         fos.write("Hello\n".getBytes());
-        Thread.sleep(150);
         fos.write("World\n".getBytes());
-        Thread.sleep(150);
         fos.write("Hello\n".getBytes());
-        Thread.sleep(150);
         fos.write("World\n".getBytes());
-        Thread.sleep(150);
         fos.write("Hello Boy\n".getBytes());
-        Thread.sleep(150);
         fos.write("World\n".getBytes());
+        fos.flush();
+        fos.close();
 
         MockEndpoint.assertIsSatisfied(context);
-
-        fos.close();
     }
 
     @Override