CAMEL-19545: Replace sleep-based synchronization in camel-stream tests

[rkdfx]

Description

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.

Target

• I checked that the commit is targeting the correct branch (Camel 4 uses the main branch)

Tracking

• If this is a large change, bug fix, or code improvement, I checked there is a JIRA issue filed for the change (usually before you start working on it).

Apache Camel coding standards and style

• I checked that each commit in the pull request has a meaningful subject line and body.

• I have run mvn clean install -DskipTests locally from root folder and I have committed all auto-generated changes.

AI-assisted contributions

• If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., Co-authored-by trailers) and the PR description identifies the AI tool used.

[rkdfx]

I'm running the mvn clean install -DskipTests locally but it's failing, it's unrelated to my changes. I'll try to clean and test again - Failed to execute goal io.github.ascopes:protobuf-maven-plugin:4.1.3:generate (default) on project camel-kserve: Generation failed - PROTOC FAILED: Protoc failed with an error. Check the build logs above to find the root cause.

[gnodet]

The goal of removing Thread.sleep() from these tests is sound, but the execution needs work — only one of the four changes actually replaces sleep with proper event-based synchronization. The others just delete sleeps and add fos.flush(), which trades one kind of fragility for another.

ScanStreamFileManualTest — drop this change

// Before:
Thread.sleep(60000);
// After:
new CountDownLatch(1).await(60, TimeUnit.SECONDS);

This test is @Disabled("For manual testing") — the 60-second pause is intentional so a human can interact with the file while the route runs. A CountDownLatch(1) that is never counted down is just Thread.sleep with extra steps: less readable, and not Awaitility. This file should be left unchanged.

ScanStreamFileTest.testScanRefreshedFile() — good ✅

await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() >= 2);

This is exactly the right pattern — event-based, deterministic, no flaky timing assumption. This is the model the other tests should follow.

ScanStreamFileTest.testScanFile() / testScanFileAlreadyWritten() — incomplete

These simply remove the sleeps and add fos.flush(). They now rely entirely on MockEndpoint.assertIsSatisfied() having an implicit 10-second internal latch timeout (MockEndpoint.waitForCompleteLatch defaults to 10s when resultWaitTime is 0). That works in practice, but it's implicit and not obvious to any reader of the test.

fos.flush() pushes bytes from Java's buffer to the OS — it does not guarantee the consumer's BufferedReader.readLine() sees the data on the very next poll cycle. The consumer polls every 200ms (scanStreamDelay=200). Without any explicit wait, these tests silently depend on the mock's internal timeout to absorb that latency. This should use Awaitility like testScanRefreshedFile does, for consistency and clarity.

ScanStreamFileWithFilterTest — correct in effect, but inconsistent

Removing the interleaved sleeps between writes is fine here (the filter only matches "Hello Boy", so processing order doesn't matter, and moving fos.close() before assertIsSatisfied is actually better). But again, no Awaitility — it silently relies on MockEndpoint's 10s timeout.

Suggestions

1. Drop the ScanStreamFileManualTest change — the sleep is intentional in a disabled manual test.
2. Keep the testScanRefreshedFile() Awaitility change as-is.
3. Add Awaitility to testScanFile(), testScanFileAlreadyWritten(), and ScanStreamFileWithFilterTest instead of just removing sleeps — e.g. await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> MockEndpoint.assertIsSatisfied(context)). This makes the synchronization explicit and consistent with the good pattern already in the PR.

[gnodet]

Thanks for addressing the previous feedback — dropping the ManualTest change and adding Awaitility everywhere is a step in the right direction.

However, MockEndpoint already has a built-in timed assertion that does exactly what the Awaitility wrapping does here:

// Current (Awaitility polling around a latch-based wait — redundant):
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> MockEndpoint.assertIsSatisfied(context));

// Better (native, latch-based, no external dependency needed):
MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);

MockEndpoint.assertIsSatisfied(CamelContext, long, TimeUnit) sets resultWaitTime on every mock in the context and uses the internal CountDownLatch to wait — it returns as soon as expectations are met or fails after the timeout. Wrapping it with Awaitility is redundant: you're polling a mechanism that already waits internally.

This applies to testScanFile(), testScanFileAlreadyWritten() in ScanStreamFileTest, and testScanFile() in ScanStreamFileWithFilterTest.

The one place where Awaitility remains the right choice is the mid-test synchronization in testScanRefreshedFile():

await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() >= 2);

MockEndpoint doesn't have a "wait until N received without asserting" API, so Awaitility is genuinely useful there. That part is good as-is.

Also, testScanRefreshedFile() still uses plain MockEndpoint.assertIsSatisfied(context) for its final assertion (no timeout override) — it should use the timed variant too, for consistency with the other tests.

[github-actions]

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 Apache Camel Committers, please review the following items:

• First-time contributors require MANUAL approval for the GitHub Actions to run
• You can use the command /component-test (camel-)component-name1 (camel-)component-name2.. to request a test from the test bot although they are normally detected and executed by CI.
• You can label PRs using skip-tests and test-dependents to fine-tune the checks executed by this PR.
• Build and test logs are available in the summary page. Only Apache Camel committers have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

[github-actions]

🧪 CI tested the following changed modules:

• components/camel-stream

---

🔬 Scalpel shadow comparison — Scalpel: 1 tested, 0 compile-only — current: 9 all tested

Maveniverse Scalpel detected 1 affected modules (current approach: 9).

Modules only in current approach (8)

• camel-jbang-mcp
• camel-jbang-plugin-mcp
• camel-jbang-plugin-route-parser
• camel-jbang-plugin-tui
• camel-jbang-plugin-validate
• camel-launcher-container
• camel-yaml-dsl-validator
• camel-yaml-dsl-validator-maven-plugin

Skip-tests mode would test 1 modules (1 direct + 0 downstream), skip tests for 0 (generated code, meta-modules)

Modules Scalpel would test (1)

• camel-stream

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (9 modules)

• Camel :: JBang :: MCP
• Camel :: JBang :: Plugin :: MCP
• Camel :: JBang :: Plugin :: Route Parser
• Camel :: JBang :: Plugin :: TUI
• Camel :: JBang :: Plugin :: Validate
• Camel :: Launcher :: Container
• Camel :: Stream
• Camel :: YAML DSL :: Validator
• Camel :: YAML DSL :: Validator Maven Plugin

---

⚙️ View full build and test results

[oscerd]

Thanks Ravi. Re-reviewing after the latest commit: both rounds of @gnodet's requested changes now look addressed, so the outstanding changes-requested appears stale.

• All Thread.sleep(...) calls are removed from the two modified test files, replaced either with the native timed assertion MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS) or, for the mid-test pre-rollover wait, await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() >= 2) — both with explicit atMost timeouts and no busy-wait.
• ScanStreamFileManualTest is correctly left untouched (its 60s sleep is an intentional @Disabled manual test).
• CI is green.

LGTM from me. @gnodet — would you mind re-reviewing, since your comments are addressed? A squash of the "addressed comments" commits at merge would be a nice-to-have.

Reviewed with Claude Code on behalf of Andrea Cosentino. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.

[gnodet]

Re-review after fix commit 4a57804: all prior findings have been fully addressed.

Prior findings status:

• ✅ ScanStreamFileManualTest changes correctly dropped (the 60s sleep is intentional in a @Disabled manual test)
• ✅ testScanFile / testScanFileAlreadyWritten / ScanStreamFileWithFilterTest.testScanFile: now use native MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS) instead of redundant Awaitility wrapper
• ✅ testScanRefreshedFile: final assertion updated to timed variant; mid-test Awaitility sync point correctly retained (no native alternative for that pattern)
• ✅ Awaitility import properly cleaned up in ScanStreamFileWithFilterTest (no longer needed) and retained in ScanStreamFileTest (still used)

CI is green. Clean implementation.

This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

Claude Code on behalf of Guillaume Nodet