fix(connectors): Harden Elasticsearch sink test readiness by ryerraguntla · Pull Request #3729 · apache/iggy

ryerraguntla · 2026-07-22T01:57:38Z

Summary

Fixes flaky Elasticsearch sink integration test where iggy-connectors
never becomes healthy because sink open() can hang on Elasticsearch HTTP
with no client timeout, while the connectors HTTP API only binds after
plugin open.
Hardens the Elasticsearch Testcontainers fixture: use 127.0.0.1, always
re-check /_cluster/health on reuse attach, self-heal a wedged
iggy-test-elasticsearch container only, and sweep stale harness indices
older than 30 minutes.
Improves harness readiness: poll connectors /health and fail fast if the
child process dies during retries; refresh the index before document-count
polls in the sink fixture.

Test plan

cargo fmt --all
cargo sort --no-format --workspace
cargo clippy -p iggy_connector_elasticsearch_sink -p integration --all-features --all-targets -- -D warnings
cargo test -p integration -- connectors::elasticsearch::elasticsearch_sink::elasticsearch_sink_stores_json_messages
Confirm docker rm -f recovery targets only iggy-test-elasticsearch
(no other Elasticsearch containers)
Optional: leave a reused iggy-test-elasticsearch running and re-run
the sink test to verify reuse + readiness path

Notes

Container force-remove uses the fixed harness name
iggy-test-elasticsearch only.
Stale index sweep is best-effort and age-gated so concurrent harness
tests sharing the reused container are not disrupted.

Add a timeout to the Elasticsearch sink transport, make connectors-runtime readiness check `/health`, and strengthen the Elasticsearch test fixture startup path with retry, cluster-health polling, and stale-index cleanup. Also refresh the sink index before counting documents so test assertions see newly indexed data sooner.

Use short-timeout probes without retries for cluster health and stale index sweep, and cap docker rm so a wedged daemon cannot stall setup.

github-actions · 2026-07-22T01:57:48Z

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

/ready - back to S-waiting-on-review after addressing feedback
/author - flip to S-waiting-on-author while you finish changes
/request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

codecov · 2026-07-22T02:09:06Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 40.21%. Comparing base (a5001d3) to head (d185f6c).
⚠️ Report is 3 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3729       +/-   ##
=============================================
- Coverage     74.34%   40.21%   -34.13%     
  Complexity      950      950               
=============================================
  Files          1303     1301        -2     
  Lines        148712   129057    -19655     
  Branches     124225   104570    -19655     
=============================================
- Hits         110562    51905    -58657     
- Misses        34673    74406    +39733     
+ Partials       3477     2746      -731

Components	Coverage Δ
Rust Core	`32.01% <100.00%> (-42.57%)`	⬇️
Java SDK	`62.64% <ø> (ø)`
C# SDK	`72.28% <ø> (ø)`
Python SDK	`92.27% <ø> (ø)`
PHP SDK	`84.52% <ø> (ø)`
Node SDK	`92.70% <ø> (ø)`
Go SDK	`43.08% <ø> (ø)`

Files with missing lines	Coverage Δ
...ore/connectors/sinks/elasticsearch_sink/src/lib.rs	`67.60% <100.00%> (+1.42%)`	⬆️

... and 415 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

ryerraguntla · 2026-07-22T17:26:01Z

/request-review @user-or-team

slbotbm

`

slbotbm · 2026-07-22T18:48:07Z

start() force-removes the shared container after any try_start() error, not only after confirming it is wedged. Transient health, port-inspection, or Docker errors can therefore remove an Elasticsearch container still used by another process. The fixed name limits which container is removed but does not provide ownership or concurrency safety. Could this recovery path verify the container is unhealthy and coordinate removal with a cross-process lock before running docker rm -f?

Let me add the logic to remove when required.

Fixed in dd2c04c

Only docker rm the shared reuse container after inspect shows a removable state, and serialize recovery with a cross-process lock so transient start flakes cannot yank a healthy instance from a peer nextest worker.

ryerraguntla · 2026-07-23T00:47:21Z

/ready

hubcio

not in this diff, same class of problem:

elasticsearch source create_client() (core/connectors/sources/elasticsearch_source/src/lib.rs:304) still builds its transport with no timeout - same open()-hang failure mode this PR fixes for the sink, so source tests keep the #3728 flake class. worth a follow-up.
mcp.rs wait_ready() still uses the pattern this PR fixed for connectors: GET /, any Ok(_) counts as ready, no process-crash fast-fail. consistency follow-up.

hubcio · 2026-07-23T09:00:41Z

+            .unwrap_or(DEFAULT_TIMEOUT_SECONDS)
+            .max(1);
+        let mut transport_builder =
+            TransportBuilder::new(conn_pool).timeout(Duration::from_secs(timeout_seconds));


this timeout is client-global in elasticsearch-rs, so it bounds bulk requests in consume() too, not just open(). that changes the failure mode: before, a bulk on degraded ES stalled until it eventually succeeded; now anything over timeout_seconds returns Err, which the runtime discards (#2927) with the offset already committed at poll (#2928) - the batch is silently dropped. mechanism is pre-existing and tracked there, but this PR is what arms it for slow-ES bulk. worth an upgrade note in the PR description: raise timeout_seconds for slow bulk workloads, and note the tradeoff until #2927/#2928 land.

separately, the comment slightly oversells the timeout as the flake fix - with the 30s default the harness readiness budget (~20s) expires first, so what fixes #3728 is the fixture readiness gate; this is the infinite-hang backstop. maybe say that instead.

hubcio · 2026-07-23T09:00:41Z

+            .config
+            .timeout_seconds
+            .unwrap_or(DEFAULT_TIMEOUT_SECONDS)
+            .max(1);


.max(1) silently turns timeout_seconds: 0 into 1s. defensible - Duration::ZERO would instant-fail every request - but undocumented, and clickhouse_sink passes 0 through unclamped, so the two siblings now disagree on the same config field. document the clamp here; clickhouse should probably get the same guard.

hubcio · 2026-07-23T09:00:41Z

+        // Prefer /health over `/` so readiness means the connectors API is up,
+        // not some other listener that happened to bind the reserved port.
+        let health_url = format!("{}/health", self.http_url());
        let client = reqwest::Client::new();


reqwest::Client::new() has no request timeout, so a black-holed GET (connection accepted, no response) hangs send() forever - the loop never gets back to the pid check above and the retry budget never fires. low risk on localhost (dead process means ECONNREFUSED, which fast-fails), but a 1-2s timeout like the fixture probe client makes both guards effective.

hubcio · 2026-07-23T09:00:42Z

+            match client.get(&health_url).send().await {
+                Ok(response) if response.status().is_success() => {
+                    let body = response.text().await.unwrap_or_default();
+                    if body.contains("\"timed_out\":true") {


?timeout=1s without a wait_for_* param makes /_cluster/health return immediately with timed_out: false, so this check never fires - and any 200 counts as ready even with cluster status red. ?wait_for_status=yellow&timeout=1s makes both the timeout param and this check do real work (and then parsing the bool beats string-matching the body).

hubcio · 2026-07-23T09:00:42Z

+}
+
+fn acquire_recovery_lock() -> Result<RecoveryLock, String> {
+    let path = std::env::temp_dir().join(RECOVERY_LOCK_FILE_NAME);


cross-process serialization only holds if every nextest process resolves the same temp_dir() - with a per-process TMPDIR (some sandbox/CI setups) each process locks its own file and recovery races silently. probably fine to accept, but worth a comment stating the assumption.

hubcio · 2026-07-23T09:00:42Z

+        .truncate(false)
+        .open(&path)
+        .map_err(|error| format!("open {}: {error}", path.display()))?;
+    file.lock()


File::lock() blocks with no acquire timeout - worst case a peer holds this through its whole recovery (two 120s container starts plus readiness waits), and since the test runtime is current-thread, the block parks the entire runtime for that long. bounded and failure-path-only, so nice-to-have: try_lock() with a deadline.

hubcio · 2026-07-23T09:00:42Z

+    ) || health == "unhealthy"
+}
+
+fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option<String> {


docker_command_stdout and force_remove_container duplicate the spawn / try_wait / deadline / kill loop, and both block the async caller (thread::sleep on a current-thread runtime). tokio::process::Command + tokio::time::timeout(dur, cmd.output()) + kill_on_drop(true) collapses both to a few lines with the same 5s/15s kill semantics - delta/fixture.rs:254 is the in-repo precedent and tokio full is already enabled here. it also removes a latent footgun: stdout is only drained after exit, so docker output over the pipe buffer (~64KB) would deadlock until the timeout kill. fine today with one-line inspect -f output, but the helper reads as generic.

hubcio · 2026-07-23T09:00:42Z

+
        for _ in 0..POLL_ATTEMPTS {
+            // Refresh so near-real-time search/count sees recently indexed docs.
+            if let Err(error) = self.refresh_index().await {


refresh_index() goes through create_http_client() (30s timeout + 3 retries) inside a 50ms poll loop - on the failure path that is up to 100 refreshes, each with retry backoff. happy path exits in a few iterations so this only inflates time-to-failure; the short-timeout probe-client pattern from container.rs would fit here too.

ryerraguntla added 2 commits July 21, 2026 14:44

fix(connectors): avoid ES fixture readiness hang

72fd3b7

Use short-timeout probes without retries for cluster health and stale index sweep, and cap docker rm so a wedged daemon cannot stall setup.

github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Jul 22, 2026

ryerraguntla changed the title ~~fix(connectors): harden Elasticsearch sink test readiness~~ fix(connectors): Harden Elasticsearch sink test readiness Jul 22, 2026

ci: retrigger pre-merge

ada822d

slbotbm requested changes Jul 22, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 22, 2026

ryerraguntla added 2 commits July 22, 2026 15:38

fix(connectors): gate ES fixture force-remove on wedged state

dd2c04c

Only docker rm the shared reuse container after inspect shows a removable state, and serialize recovery with a cross-process lock so transient start flakes cannot yank a healthy instance from a peer nextest worker.

fix(connectors): silence unused RecoveryLock field for clippy

d185f6c

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Jul 23, 2026

hubcio requested changes Jul 23, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 23, 2026

Uh oh!

Conversation

ryerraguntla commented Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Test plan

Notes

Uh oh!

github-actions Bot commented Jul 22, 2026

Uh oh!

codecov Bot commented Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

ryerraguntla commented Jul 22, 2026

Uh oh!

slbotbm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla commented Jul 23, 2026

Uh oh!

hubcio left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ryerraguntla commented Jul 22, 2026 •

edited

Loading

codecov Bot commented Jul 22, 2026 •

edited

Loading

ryerraguntla Jul 22, 2026 •

edited

Loading

ryerraguntla Jul 22, 2026 •

edited

Loading