From 2d221c12f53fb3ad6f37b5e1b8c232e9c673f84b Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Tue, 21 Jul 2026 14:44:33 -0400 Subject: [PATCH 1/8] Harden Elasticsearch connector tests 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. --- .../sinks/elasticsearch_sink/src/lib.rs | 13 +- .../src/harness/handle/connectors_runtime.rs | 24 ++- .../fixtures/elasticsearch/container.rs | 182 +++++++++++++++++- .../connectors/fixtures/elasticsearch/sink.rs | 24 ++- .../fixtures/elasticsearch/source.rs | 2 - 5 files changed, 229 insertions(+), 16 deletions(-) diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs index d4a8cb99d3..8895e35327 100644 --- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs +++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs @@ -31,11 +31,14 @@ use secrecy::{ExposeSecret, SecretString}; use serde::{Deserialize, Serialize}; use serde_json::json; use simd_json::{OwnedValue, prelude::*}; +use std::time::Duration; use tokio::sync::Mutex; use tracing::{info, warn}; sink_connector!(ElasticsearchSink); +const DEFAULT_TIMEOUT_SECONDS: u64 = 30; + #[derive(Debug)] struct State { invocations_count: usize, @@ -83,7 +86,15 @@ impl ElasticsearchSink { .map_err(|error| Error::Connection(format!("Invalid Elasticsearch URL: {error}")))?; let conn_pool = elasticsearch::http::transport::SingleNodeConnectionPool::new(url); - let mut transport_builder = TransportBuilder::new(conn_pool); + // elasticsearch-rs defaults to no timeout; without this, a hung ES + // connection blocks FFI open() and the connectors HTTP health never binds. + let timeout_seconds = self + .config + .timeout_seconds + .unwrap_or(DEFAULT_TIMEOUT_SECONDS) + .max(1); + let mut transport_builder = + TransportBuilder::new(conn_pool).timeout(Duration::from_secs(timeout_seconds)); if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { let credentials = diff --git a/core/integration/src/harness/handle/connectors_runtime.rs b/core/integration/src/harness/handle/connectors_runtime.rs index f86eb91436..2ee548dab8 100644 --- a/core/integration/src/harness/handle/connectors_runtime.rs +++ b/core/integration/src/harness/handle/connectors_runtime.rs @@ -243,19 +243,33 @@ impl IggyServerDependent for ConnectorsRuntimeHandle { } async fn wait_ready(&mut self) -> Result<(), TestBinaryError> { - let http_address = self.http_url(); + // 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(); for retry in 0..common::DEFAULT_HEALTH_CHECK_RETRIES { - match client.get(&http_address).send().await { - Ok(_) => { + if let Some(pid) = self.pid() + && !common::is_process_alive(pid) + { + let (stdout, stderr) = self.collect_logs(); + return Err(TestBinaryError::ProcessCrashed { + binary: "iggy-connectors".to_string(), + exit_code: None, + stdout, + stderr, + }); + } + + match client.get(&health_url).send().await { + Ok(response) if response.status().is_success() => { return Ok(()); } - Err(_) => { + Ok(_) | Err(_) => { if retry == common::DEFAULT_HEALTH_CHECK_RETRIES - 1 { return Err(TestBinaryError::HealthCheckFailed { binary: "iggy-connectors".to_string(), - address: http_address, + address: health_url, retries: common::DEFAULT_HEALTH_CHECK_RETRIES, }); } diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index 5749b18096..8d64b768b5 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -26,7 +26,7 @@ use testcontainers_modules::testcontainers::runners::AsyncRunner; use testcontainers_modules::testcontainers::{ ContainerAsync, GenericImage, ImageExt, ReuseDirective, }; -use tracing::info; +use tracing::{info, warn}; const ELASTICSEARCH_IMAGE: &str = "docker.io/library/elasticsearch"; const ELASTICSEARCH_TAG: &str = "9.3.0"; @@ -37,6 +37,13 @@ const ELASTICSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health"; // name. Per-test isolation comes from a unique index per fixture, not a fresh // container. const ELASTICSEARCH_CONTAINER_NAME: &str = "iggy-test-elasticsearch"; +const CLUSTER_READY_ATTEMPTS: usize = 60; +const CLUSTER_READY_INTERVAL_MS: u64 = 500; +// Indices from prior runs older than this are leftovers: a live concurrent +// test's index is seconds old, so age-based sweeping never races other tests +// sharing the reused container. +const STALE_INDEX_MAX_AGE_MS: u128 = 30 * 60 * 1000; +const STALE_INDEX_PATTERNS: &str = "iggy_messages_*,test_documents_*"; pub const DEFAULT_TEST_STREAM: &str = "test_stream"; pub const DEFAULT_TEST_TOPIC: &str = "test_topic"; @@ -101,6 +108,23 @@ pub struct ElasticsearchContainer { impl ElasticsearchContainer { pub async fn start() -> Result { + match Self::try_start().await { + Ok(started) => Ok(started), + Err(first_error) => { + // A reused container can be wedged (running but unhealthy: + // OOM-killed JVM, corrupted data dir, dead port mapping). + // Remove it and retry once with a fresh container instead of + // failing every test until someone cleans up manually. + warn!( + "Elasticsearch container unusable, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}" + ); + force_remove_container(); + Self::try_start().await + } + } + } + + async fn try_start() -> Result { let container = GenericImage::new(ELASTICSEARCH_IMAGE, ELASTICSEARCH_TAG) .with_exposed_port(ELASTICSEARCH_PORT.tcp()) .with_wait_for(WaitFor::http( @@ -137,14 +161,166 @@ impl ElasticsearchContainer { message: "No mapping for Elasticsearch port".to_string(), })?; - let base_url = format!("http://localhost:{mapped_port}"); + // Prefer IPv4 loopback: Docker publishes 0.0.0.0:HOST→9200. `localhost` + // can resolve to ::1 first on macOS and black-hole the elasticsearch-rs + // client while the fixture's reqwest client still looks healthy. + let base_url = format!("http://127.0.0.1:{mapped_port}"); info!("Elasticsearch container available at {base_url}"); - Ok(Self { + let started = Self { container, base_url, + }; + // ReuseDirective::Always can attach to a days-old container without + // re-running HttpWaitStrategy; verify cluster health on every setup. + started.wait_until_ready().await?; + started.sweep_stale_indices().await; + Ok(started) + } + + async fn wait_until_ready(&self) -> Result<(), TestBinaryError> { + let client = create_http_client(); + let health_url = format!("{}{ELASTICSEARCH_HEALTH_ENDPOINT}", self.base_url); + let mut last_error = String::from("no attempts made"); + + for attempt in 1..=CLUSTER_READY_ATTEMPTS { + 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") { + last_error = format!( + "cluster health timed out on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {body}" + ); + } else { + info!("Elasticsearch cluster ready at {}", self.base_url); + return Ok(()); + } + } + Ok(response) => { + last_error = format!( + "cluster health status {} on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}", + response.status() + ); + } + Err(error) => { + last_error = format!( + "cluster health request failed on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {error}" + ); + } + } + tokio::time::sleep(std::time::Duration::from_millis(CLUSTER_READY_INTERVAL_MS)).await; + } + + Err(TestBinaryError::FixtureSetup { + fixture_type: "ElasticsearchContainer".to_string(), + message: format!( + "Elasticsearch at {} not ready after {CLUSTER_READY_ATTEMPTS} attempts: {last_error}", + self.base_url + ), }) } + + /// Delete leftover test indices (empty or partially filled) from previous + /// runs so accumulated shards do not degrade the reused container. Only + /// indices older than [`STALE_INDEX_MAX_AGE_MS`] are removed, which keeps + /// the sweep safe against tests running concurrently in other processes. + /// Best-effort: failures are logged, never fail the fixture. + async fn sweep_stale_indices(&self) { + #[derive(Deserialize)] + struct CatIndexEntry { + index: String, + #[serde(rename = "creation.date")] + creation_date: Option, + } + + let client = create_http_client(); + let cat_url = format!( + "{}/_cat/indices/{STALE_INDEX_PATTERNS}?format=json&h=index,creation.date", + self.base_url + ); + + let entries = match client.get(&cat_url).send().await { + Ok(response) if response.status().is_success() => { + match response.json::>().await { + Ok(entries) => entries, + Err(error) => { + warn!("Skipping stale index sweep, unparsable _cat response: {error}"); + return; + } + } + } + Ok(response) => { + warn!( + "Skipping stale index sweep, _cat/indices returned {}", + response.status() + ); + return; + } + Err(error) => { + warn!("Skipping stale index sweep, _cat/indices failed: {error}"); + return; + } + }; + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis()) + .unwrap_or(0); + + let stale: Vec = entries + .into_iter() + .filter_map(|entry| { + let created_ms = entry.creation_date.as_deref()?.parse::().ok()?; + (now_ms.saturating_sub(created_ms) > STALE_INDEX_MAX_AGE_MS).then_some(entry.index) + }) + .collect(); + + if stale.is_empty() { + return; + } + + // Chunked so the URL stays well under limits with many leftovers. + for chunk in stale.chunks(20) { + let delete_url = format!("{}/{}", self.base_url, chunk.join(",")); + match client.delete(&delete_url).send().await { + Ok(response) if response.status().is_success() => { + info!("Deleted {} stale Elasticsearch test indices", chunk.len()); + } + Ok(response) => { + warn!( + "Failed to delete stale Elasticsearch indices, status {}", + response.status() + ); + } + Err(error) => { + warn!("Failed to delete stale Elasticsearch indices: {error}"); + } + } + } + } +} + +/// Remove the shared reuse container so the next start creates a fresh one. +/// Uses the Docker CLI directly: testcontainers offers no "remove by name" API +/// and the wedged container was created by an earlier process anyway. +fn force_remove_container() { + match std::process::Command::new("docker") + .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME]) + .output() + { + Ok(output) if output.status.success() => { + info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); + } + Ok(output) => { + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + Err(error) => { + warn!("Failed to invoke docker rm for '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); + } + } } pub fn create_http_client() -> HttpClient { diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs index 10462a2028..03313db377 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs @@ -61,22 +61,38 @@ impl ElasticsearchSinkFixture { &self, expected_count: usize, ) -> Result { + let mut last_error: Option = None; + for _ in 0..POLL_ATTEMPTS { + // Refresh so near-real-time search/count sees recently indexed docs. + if let Err(error) = self.refresh_index().await { + last_error = Some(error); + sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; + continue; + } + match self.count_documents(&self.index).await { Ok(count) if count >= expected_count => { info!("Found {count} documents in Elasticsearch (expected {expected_count})"); return Ok(count); } - Ok(_) => {} - Err(_) => {} + Ok(_) => { + last_error = None; + } + Err(error) => { + last_error = Some(error); + } } sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; } let final_count = self.count_documents(&self.index).await.unwrap_or(0); + let detail = last_error + .map(|error| format!("; last error: {error}")) + .unwrap_or_default(); Err(TestBinaryError::InvalidState { message: format!( - "Expected at least {expected_count} documents, found {final_count} after {POLL_ATTEMPTS} attempts" + "Expected at least {expected_count} documents, found {final_count} after {POLL_ATTEMPTS} attempts{detail}" ), }) } @@ -97,8 +113,6 @@ impl TestFixture for ElasticsearchSinkFixture { let http_client = create_http_client(); let index = format!("{SINK_INDEX_PREFIX}_{}", Uuid::new_v4().simple()); - // Container startup already waits for /_cluster/health to return 200 - // via HttpWaitStrategy, so no additional health check is needed. Ok(Self { container, http_client, diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs index e6c42096bc..2368fce3d5 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs @@ -101,8 +101,6 @@ impl TestFixture for ElasticsearchSourceFixture { let http_client = create_http_client(); let index = format!("{TEST_INDEX_PREFIX}_{}", Uuid::new_v4().simple()); - // Container startup already waits for /_cluster/health to return 200 - // via HttpWaitStrategy, so no additional health check is needed. Ok(Self { container, http_client, From 72fd3b72a89c3b7c24d31582f51e1be82b2bb4e3 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Tue, 21 Jul 2026 15:57:44 -0400 Subject: [PATCH 2/8] fix(connectors): avoid ES fixture readiness hang Use short-timeout probes without retries for cluster health and stale index sweep, and cap docker rm so a wedged daemon cannot stall setup. --- .../fixtures/elasticsearch/container.rs | 89 +++++++++++++++---- 1 file changed, 73 insertions(+), 16 deletions(-) diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index 8d64b768b5..bedfea91d3 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -20,6 +20,9 @@ use reqwest_middleware::ClientWithMiddleware as HttpClient; use reqwest_retry::RetryTransientMiddleware; use reqwest_retry::policies::ExponentialBackoff; use serde::Deserialize; +use std::io::Read; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy; use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; use testcontainers_modules::testcontainers::runners::AsyncRunner; @@ -37,8 +40,12 @@ const ELASTICSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health"; // name. Per-test isolation comes from a unique index per fixture, not a fresh // container. const ELASTICSEARCH_CONTAINER_NAME: &str = "iggy-test-elasticsearch"; -const CLUSTER_READY_ATTEMPTS: usize = 60; -const CLUSTER_READY_INTERVAL_MS: u64 = 500; +// Short probe timeouts: create_http_client() uses 30s + retries and must not +// be used for readiness. One hung attempt there looks like a 60s+ test hang. +const CLUSTER_READY_ATTEMPTS: usize = 40; +const CLUSTER_READY_INTERVAL_MS: u64 = 250; +const CLUSTER_READY_REQUEST_TIMEOUT_MS: u64 = 2_000; +const DOCKER_RM_TIMEOUT_SECS: u64 = 15; // Indices from prior runs older than this are leftovers: a live concurrent // test's index is seconds old, so age-based sweeping never races other tests // sharing the reused container. @@ -179,8 +186,23 @@ impl ElasticsearchContainer { } async fn wait_until_ready(&self) -> Result<(), TestBinaryError> { - let client = create_http_client(); - let health_url = format!("{}{ELASTICSEARCH_HEALTH_ENDPOINT}", self.base_url); + // Dedicated probe client: short timeout, no retry middleware. The shared + // create_http_client() (30s + 3 retries) turns one black-holed request + // into a multi-minute hang that looks like the test is stuck. + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_millis( + CLUSTER_READY_REQUEST_TIMEOUT_MS, + )) + .build() + .map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "ElasticsearchContainer".to_string(), + message: format!("Failed to build readiness HTTP client: {error}"), + })?; + // timeout=1s keeps ES from holding the request when the cluster is slow. + let health_url = format!( + "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?timeout=1s", + self.base_url + ); let mut last_error = String::from("no attempts made"); for attempt in 1..=CLUSTER_READY_ATTEMPTS { @@ -233,7 +255,14 @@ impl ElasticsearchContainer { creation_date: Option, } - let client = create_http_client(); + // Short timeout, no retries: sweep is best-effort and must not stall setup. + let Ok(client) = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build() + else { + warn!("Skipping stale index sweep, failed to build HTTP client"); + return; + }; let cat_url = format!( "{}/_cat/indices/{STALE_INDEX_PATTERNS}?format=json&h=index,creation.date", self.base_url @@ -304,21 +333,49 @@ impl ElasticsearchContainer { /// Uses the Docker CLI directly: testcontainers offers no "remove by name" API /// and the wedged container was created by an earlier process anyway. fn force_remove_container() { - match std::process::Command::new("docker") + let mut child = match Command::new("docker") .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME]) - .output() + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() { - Ok(output) if output.status.success() => { - info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); - } - Ok(output) => { - warn!( - "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed: {}", - String::from_utf8_lossy(&output.stderr) - ); - } + Ok(child) => child, Err(error) => { warn!("Failed to invoke docker rm for '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); + return; + } + }; + + let deadline = Instant::now() + Duration::from_secs(DOCKER_RM_TIMEOUT_SECS); + loop { + match child.try_wait() { + Ok(Some(status)) if status.success() => { + info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); + return; + } + Ok(Some(status)) => { + let mut stderr = String::new(); + if let Some(mut pipe) = child.stderr.take() { + let _ = pipe.read_to_string(&mut stderr); + } + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed (exit {status}): {stderr}" + ); + return; + } + Ok(None) if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} timed out after {DOCKER_RM_TIMEOUT_SECS}s" + ); + return; + } + Ok(None) => std::thread::sleep(Duration::from_millis(100)), + Err(error) => { + warn!("Failed waiting for docker rm '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); + return; + } } } } From ada822d1440b703f0b42dcbc1a1ecf7a7340fffc Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Wed, 22 Jul 2026 13:12:52 -0400 Subject: [PATCH 3/8] ci: retrigger pre-merge From dd2c04ce743c78260a26d8992a02138c00ff8fc3 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Wed, 22 Jul 2026 15:38:04 -0400 Subject: [PATCH 4/8] fix(connectors): gate ES fixture force-remove on wedged state 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. --- .../fixtures/elasticsearch/container.rs | 129 +++++++++++++++++- 1 file changed, 123 insertions(+), 6 deletions(-) diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index bedfea91d3..838680813a 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -20,6 +20,7 @@ use reqwest_middleware::ClientWithMiddleware as HttpClient; use reqwest_retry::RetryTransientMiddleware; use reqwest_retry::policies::ExponentialBackoff; use serde::Deserialize; +use std::fs::{File, OpenOptions}; use std::io::Read; use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; @@ -46,6 +47,10 @@ const CLUSTER_READY_ATTEMPTS: usize = 40; const CLUSTER_READY_INTERVAL_MS: u64 = 250; const CLUSTER_READY_REQUEST_TIMEOUT_MS: u64 = 2_000; const DOCKER_RM_TIMEOUT_SECS: u64 = 15; +const DOCKER_INSPECT_TIMEOUT_SECS: u64 = 5; +// Serializes inspect+rm recovery across nextest processes sharing the reused +// container. Held only on the failure path. +const RECOVERY_LOCK_FILE_NAME: &str = "iggy-test-elasticsearch-recovery.lock"; // Indices from prior runs older than this are leftovers: a live concurrent // test's index is seconds old, so age-based sweeping never races other tests // sharing the reused container. @@ -118,15 +123,37 @@ impl ElasticsearchContainer { match Self::try_start().await { Ok(started) => Ok(started), Err(first_error) => { - // A reused container can be wedged (running but unhealthy: - // OOM-killed JVM, corrupted data dir, dead port mapping). - // Remove it and retry once with a fresh container instead of - // failing every test until someone cleans up manually. + // Shared Always-reuse container: never `docker rm -f` on a + // transient start/port/health flake. Another nextest worker may + // still be using a healthy instance. Recover only when inspect + // shows a removable wedged state, under a cross-process lock. + let recovery_lock = match acquire_recovery_lock() { + Ok(lock) => lock, + Err(error) => { + warn!("Skipping Elasticsearch recovery, could not take lock: {error}"); + return Err(first_error); + } + }; + + if let Ok(started) = Self::try_start().await { + drop(recovery_lock); + return Ok(started); + } + + if !container_is_removable_wedged() { + warn!( + "Elasticsearch start failed and '{ELASTICSEARCH_CONTAINER_NAME}' is not in a removable wedged state; leaving it in place: {first_error}" + ); + return Err(first_error); + } + warn!( - "Elasticsearch container unusable, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}" + "Elasticsearch container wedged, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}" ); force_remove_container(); - Self::try_start().await + let retry_result = Self::try_start().await; + drop(recovery_lock); + retry_result } } } @@ -329,9 +356,99 @@ impl ElasticsearchContainer { } } +/// Cross-process advisory lock for inspect+rm recovery of the shared reuse +/// container. Dropping the file releases the lock (including on process crash). +struct RecoveryLock(File); + +fn acquire_recovery_lock() -> Result { + let path = std::env::temp_dir().join(RECOVERY_LOCK_FILE_NAME); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|error| format!("open {}: {error}", path.display()))?; + file.lock() + .map_err(|error| format!("lock {}: {error}", path.display()))?; + Ok(RecoveryLock(file)) +} + +/// True only when Docker says the shared container is in a state safe to +/// force-remove without yanking a healthy instance from a peer process. +/// +/// Removable: exited/dead/created/paused/restarting, or Docker healthcheck +/// `unhealthy`. A plain `running` container (even if ES HTTP is flaky) is left +/// alone: readiness flakes must not `docker rm -f` a shared Always-reuse box. +fn container_is_removable_wedged() -> bool { + let inspect_format = "{{.State.Status}}|{{if .State.Health}}{{.State.Health.Status}}{{end}}"; + let Some(stdout) = docker_command_stdout( + &[ + "inspect", + "-f", + inspect_format, + ELASTICSEARCH_CONTAINER_NAME, + ], + DOCKER_INSPECT_TIMEOUT_SECS, + ) else { + return false; + }; + + let mut parts = stdout.trim().split('|'); + let status = parts.next().unwrap_or("").trim(); + let health = parts.next().unwrap_or("").trim(); + + matches!( + status, + "exited" | "dead" | "created" | "paused" | "restarting" + ) || health == "unhealthy" +} + +fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option { + let mut child = match Command::new("docker") + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + { + Ok(child) => child, + Err(error) => { + warn!("Failed to invoke docker {}: {error}", args.join(" ")); + return None; + } + }; + + let deadline = Instant::now() + Duration::from_secs(timeout_secs); + loop { + match child.try_wait() { + Ok(Some(status)) if status.success() => { + let mut stdout = String::new(); + if let Some(mut pipe) = child.stdout.take() { + let _ = pipe.read_to_string(&mut stdout); + } + return Some(stdout); + } + Ok(Some(_)) => return None, + Ok(None) if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + warn!("docker {} timed out after {timeout_secs}s", args.join(" ")); + return None; + } + Ok(None) => std::thread::sleep(Duration::from_millis(100)), + Err(error) => { + warn!("Failed waiting for docker {}: {error}", args.join(" ")); + return None; + } + } + } +} + /// Remove the shared reuse container so the next start creates a fresh one. /// Uses the Docker CLI directly: testcontainers offers no "remove by name" API /// and the wedged container was created by an earlier process anyway. +/// Caller must hold [`RecoveryLock`] and have confirmed +/// [`container_is_removable_wedged`]. fn force_remove_container() { let mut child = match Command::new("docker") .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME]) From d185f6c015f152928887b35997be6867602c7c47 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Wed, 22 Jul 2026 16:27:08 -0400 Subject: [PATCH 5/8] fix(connectors): silence unused RecoveryLock field for clippy --- .../tests/connectors/fixtures/elasticsearch/container.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index 838680813a..78ce3ff452 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -358,7 +358,9 @@ impl ElasticsearchContainer { /// Cross-process advisory lock for inspect+rm recovery of the shared reuse /// container. Dropping the file releases the lock (including on process crash). -struct RecoveryLock(File); +struct RecoveryLock { + _file: File, +} fn acquire_recovery_lock() -> Result { let path = std::env::temp_dir().join(RECOVERY_LOCK_FILE_NAME); @@ -371,7 +373,7 @@ fn acquire_recovery_lock() -> Result { .map_err(|error| format!("open {}: {error}", path.display()))?; file.lock() .map_err(|error| format!("lock {}: {error}", path.display()))?; - Ok(RecoveryLock(file)) + Ok(RecoveryLock { _file: file }) } /// True only when Docker says the shared container is in a state safe to From 62933fc2951097d7107e25bfe48685a7f5f706f6 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Fri, 24 Jul 2026 13:46:35 -0400 Subject: [PATCH 6/8] fix(connectors): bound ES harness probes against hangs short-timeout health clients, wait_for_status , cluster readiness, async try_lock recovery, and timeout docs clarifying client-global clamp vs #3728 flake fix. --- .../sinks/elasticsearch_sink/README.md | 2 +- .../sinks/elasticsearch_sink/config.toml | 2 + .../sinks/elasticsearch_sink/src/lib.rs | 11 +- .../src/harness/handle/connectors_runtime.rs | 10 +- .../fixtures/elasticsearch/container.rs | 197 ++++++++++-------- .../connectors/fixtures/elasticsearch/sink.rs | 30 ++- 6 files changed, 155 insertions(+), 97 deletions(-) diff --git a/core/connectors/sinks/elasticsearch_sink/README.md b/core/connectors/sinks/elasticsearch_sink/README.md index 905fbf496b..0ba651b941 100644 --- a/core/connectors/sinks/elasticsearch_sink/README.md +++ b/core/connectors/sinks/elasticsearch_sink/README.md @@ -8,7 +8,7 @@ A sink connector that consumes messages from Iggy streams and indexes them to El - `index`: Target index name - `username/password`: Optional authentication credentials - `batch_size`: Bulk indexing batch size (default: 100) -- `timeout_seconds`: Request timeout (default: 30s) +- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed (see #2927/#2928) - `create_index_if_not_exists`: Automatically create index (default: true) - `index_mapping`: Index mapping configuration diff --git a/core/connectors/sinks/elasticsearch_sink/config.toml b/core/connectors/sinks/elasticsearch_sink/config.toml index 171e0eec60..75c6ed4748 100644 --- a/core/connectors/sinks/elasticsearch_sink/config.toml +++ b/core/connectors/sinks/elasticsearch_sink/config.toml @@ -36,4 +36,6 @@ url = "http://localhost:9200" index = "iggy_messages" create_index_if_not_exists = true batch_size = 100 +# Client-wide timeout for open() and bulk consume(). 0 is clamped to 1s. +# Raise for slow bulk; timed-out consume drops the batch after poll commit (#2927/#2928). timeout_seconds = 30 diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs index 8895e35327..372fd2b8f6 100644 --- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs +++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs @@ -54,6 +54,11 @@ pub struct ElasticsearchSinkConfig { #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] pub password: Option, pub batch_size: Option, + /// Client-wide HTTP timeout for open() and bulk consume(). Values of `0` + /// are clamped to 1s (a zero duration would fail every request immediately). + /// Raise this for slow bulk workloads; until runtime ack/retry (#2927/#2928), + /// a timeout on consume drops the batch after the poll offset is already + /// committed. pub timeout_seconds: Option, pub create_index_if_not_exists: Option, pub index_mapping: Option, @@ -86,8 +91,10 @@ impl ElasticsearchSink { .map_err(|error| Error::Connection(format!("Invalid Elasticsearch URL: {error}")))?; let conn_pool = elasticsearch::http::transport::SingleNodeConnectionPool::new(url); - // elasticsearch-rs defaults to no timeout; without this, a hung ES - // connection blocks FFI open() and the connectors HTTP health never binds. + // elasticsearch-rs defaults to no timeout. This client-global timeout is + // an infinite-hang backstop for open() and for bulk consume() — not the + // primary #3728 flake fix (that is the harness readiness gate, which + // expires sooner than the 30s default). Values of 0 clamp to 1s. let timeout_seconds = self .config .timeout_seconds diff --git a/core/integration/src/harness/handle/connectors_runtime.rs b/core/integration/src/harness/handle/connectors_runtime.rs index 2ee548dab8..1118f55432 100644 --- a/core/integration/src/harness/handle/connectors_runtime.rs +++ b/core/integration/src/harness/handle/connectors_runtime.rs @@ -246,7 +246,15 @@ impl IggyServerDependent for ConnectorsRuntimeHandle { // 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(); + // Bound each probe so a black-holed TCP accept (no HTTP response) cannot + // stall send() past the pid-crash / retry budget. Localhost dead process + // usually ECONNREFUSED, but a short timeout keeps both guards effective. + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to build connectors health client: {error}"), + })?; for retry in 0..common::DEFAULT_HEALTH_CHECK_RETRIES { if let Some(pid) = self.pid() diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index 78ce3ff452..a0ea1a13ea 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -20,9 +20,8 @@ use reqwest_middleware::ClientWithMiddleware as HttpClient; use reqwest_retry::RetryTransientMiddleware; use reqwest_retry::policies::ExponentialBackoff; use serde::Deserialize; -use std::fs::{File, OpenOptions}; -use std::io::Read; -use std::process::{Command, Stdio}; +use std::fs::{File, OpenOptions, TryLockError}; +use std::process::Stdio; use std::time::{Duration, Instant}; use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy; use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; @@ -51,6 +50,10 @@ const DOCKER_INSPECT_TIMEOUT_SECS: u64 = 5; // Serializes inspect+rm recovery across nextest processes sharing the reused // container. Held only on the failure path. const RECOVERY_LOCK_FILE_NAME: &str = "iggy-test-elasticsearch-recovery.lock"; +// Bound lock wait so a peer mid-recovery cannot park a current-thread runtime +// for two full container startups with no upper limit. +const RECOVERY_LOCK_TIMEOUT_SECS: u64 = 30; +const RECOVERY_LOCK_POLL_MS: u64 = 50; // Indices from prior runs older than this are leftovers: a live concurrent // test's index is seconds old, so age-based sweeping never races other tests // sharing the reused container. @@ -127,7 +130,7 @@ impl ElasticsearchContainer { // transient start/port/health flake. Another nextest worker may // still be using a healthy instance. Recover only when inspect // shows a removable wedged state, under a cross-process lock. - let recovery_lock = match acquire_recovery_lock() { + let recovery_lock = match acquire_recovery_lock().await { Ok(lock) => lock, Err(error) => { warn!("Skipping Elasticsearch recovery, could not take lock: {error}"); @@ -140,7 +143,7 @@ impl ElasticsearchContainer { return Ok(started); } - if !container_is_removable_wedged() { + if !container_is_removable_wedged().await { warn!( "Elasticsearch start failed and '{ELASTICSEARCH_CONTAINER_NAME}' is not in a removable wedged state; leaving it in place: {first_error}" ); @@ -150,7 +153,7 @@ impl ElasticsearchContainer { warn!( "Elasticsearch container wedged, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}" ); - force_remove_container(); + force_remove_container().await; let retry_result = Self::try_start().await; drop(recovery_lock); retry_result @@ -213,6 +216,12 @@ impl ElasticsearchContainer { } async fn wait_until_ready(&self) -> Result<(), TestBinaryError> { + #[derive(Deserialize)] + struct ClusterHealth { + timed_out: bool, + status: String, + } + // Dedicated probe client: short timeout, no retry middleware. The shared // create_http_client() (30s + 3 retries) turns one black-holed request // into a multi-minute hang that looks like the test is stuck. @@ -225,9 +234,11 @@ impl ElasticsearchContainer { fixture_type: "ElasticsearchContainer".to_string(), message: format!("Failed to build readiness HTTP client: {error}"), })?; - // timeout=1s keeps ES from holding the request when the cluster is slow. + // wait_for_status=yellow makes ES block until yellow/green or timeout=1s + // elapses; without wait_for_*, timeout is ignored and red still returns + // 200 with timed_out=false. let health_url = format!( - "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?timeout=1s", + "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?wait_for_status=yellow&timeout=1s", self.base_url ); let mut last_error = String::from("no attempts made"); @@ -235,14 +246,25 @@ impl ElasticsearchContainer { for attempt in 1..=CLUSTER_READY_ATTEMPTS { 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") { - last_error = format!( - "cluster health timed out on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {body}" - ); - } else { - info!("Elasticsearch cluster ready at {}", self.base_url); - return Ok(()); + match response.json::().await { + Ok(health) if !health.timed_out => { + info!( + "Elasticsearch cluster ready at {} (status={})", + self.base_url, health.status + ); + return Ok(()); + } + Ok(health) => { + last_error = format!( + "cluster health timed out on attempt {attempt}/{CLUSTER_READY_ATTEMPTS} (status={})", + health.status + ); + } + Err(error) => { + last_error = format!( + "cluster health body unparsable on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {error}" + ); + } } } Ok(response) => { @@ -358,11 +380,15 @@ impl ElasticsearchContainer { /// Cross-process advisory lock for inspect+rm recovery of the shared reuse /// container. Dropping the file releases the lock (including on process crash). +/// +/// Assumes every nextest process shares the same `std::env::temp_dir()` (and +/// thus the same lock file). A per-process `TMPDIR` makes each worker lock its +/// own path and recovery can race silently. struct RecoveryLock { _file: File, } -fn acquire_recovery_lock() -> Result { +async fn acquire_recovery_lock() -> Result { let path = std::env::temp_dir().join(RECOVERY_LOCK_FILE_NAME); let file = OpenOptions::new() .read(true) @@ -371,9 +397,25 @@ fn acquire_recovery_lock() -> Result { .truncate(false) .open(&path) .map_err(|error| format!("open {}: {error}", path.display()))?; - file.lock() - .map_err(|error| format!("lock {}: {error}", path.display()))?; - Ok(RecoveryLock { _file: file }) + + let deadline = Instant::now() + Duration::from_secs(RECOVERY_LOCK_TIMEOUT_SECS); + loop { + match file.try_lock() { + Ok(()) => return Ok(RecoveryLock { _file: file }), + Err(TryLockError::WouldBlock) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(RECOVERY_LOCK_POLL_MS)).await; + } + Err(TryLockError::WouldBlock) => { + return Err(format!( + "timed out after {RECOVERY_LOCK_TIMEOUT_SECS}s waiting for {}", + path.display() + )); + } + Err(TryLockError::Error(error)) => { + return Err(format!("lock {}: {error}", path.display())); + } + } + } } /// True only when Docker says the shared container is in a state safe to @@ -382,7 +424,7 @@ fn acquire_recovery_lock() -> Result { /// Removable: exited/dead/created/paused/restarting, or Docker healthcheck /// `unhealthy`. A plain `running` container (even if ES HTTP is flaky) is left /// alone: readiness flakes must not `docker rm -f` a shared Always-reuse box. -fn container_is_removable_wedged() -> bool { +async fn container_is_removable_wedged() -> bool { let inspect_format = "{{.State.Status}}|{{if .State.Health}}{{.State.Health.Status}}{{end}}"; let Some(stdout) = docker_command_stdout( &[ @@ -392,7 +434,9 @@ fn container_is_removable_wedged() -> bool { ELASTICSEARCH_CONTAINER_NAME, ], DOCKER_INSPECT_TIMEOUT_SECS, - ) else { + ) + .await + else { return false; }; @@ -406,42 +450,26 @@ fn container_is_removable_wedged() -> bool { ) || health == "unhealthy" } -fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option { - let mut child = match Command::new("docker") +async fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option { + let mut command = tokio::process::Command::new("docker"); + command .args(args) .stdout(Stdio::piped()) .stderr(Stdio::null()) - .spawn() - { - Ok(child) => child, - Err(error) => { + .kill_on_drop(true); + + match tokio::time::timeout(Duration::from_secs(timeout_secs), command.output()).await { + Ok(Ok(output)) if output.status.success() => { + Some(String::from_utf8_lossy(&output.stdout).into_owned()) + } + Ok(Ok(_)) => None, + Ok(Err(error)) => { warn!("Failed to invoke docker {}: {error}", args.join(" ")); - return None; + None } - }; - - let deadline = Instant::now() + Duration::from_secs(timeout_secs); - loop { - match child.try_wait() { - Ok(Some(status)) if status.success() => { - let mut stdout = String::new(); - if let Some(mut pipe) = child.stdout.take() { - let _ = pipe.read_to_string(&mut stdout); - } - return Some(stdout); - } - Ok(Some(_)) => return None, - Ok(None) if Instant::now() >= deadline => { - let _ = child.kill(); - let _ = child.wait(); - warn!("docker {} timed out after {timeout_secs}s", args.join(" ")); - return None; - } - Ok(None) => std::thread::sleep(Duration::from_millis(100)), - Err(error) => { - warn!("Failed waiting for docker {}: {error}", args.join(" ")); - return None; - } + Err(_) => { + warn!("docker {} timed out after {timeout_secs}s", args.join(" ")); + None } } } @@ -451,50 +479,37 @@ fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option { /// and the wedged container was created by an earlier process anyway. /// Caller must hold [`RecoveryLock`] and have confirmed /// [`container_is_removable_wedged`]. -fn force_remove_container() { - let mut child = match Command::new("docker") +async fn force_remove_container() { + let mut command = tokio::process::Command::new("docker"); + command .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME]) .stdout(Stdio::null()) .stderr(Stdio::piped()) - .spawn() + .kill_on_drop(true); + + match tokio::time::timeout( + Duration::from_secs(DOCKER_RM_TIMEOUT_SECS), + command.output(), + ) + .await { - Ok(child) => child, - Err(error) => { + Ok(Ok(output)) if output.status.success() => { + info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); + } + Ok(Ok(output)) => { + let stderr = String::from_utf8_lossy(&output.stderr); + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed (exit {}): {stderr}", + output.status + ); + } + Ok(Err(error)) => { warn!("Failed to invoke docker rm for '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); - return; } - }; - - let deadline = Instant::now() + Duration::from_secs(DOCKER_RM_TIMEOUT_SECS); - loop { - match child.try_wait() { - Ok(Some(status)) if status.success() => { - info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); - return; - } - Ok(Some(status)) => { - let mut stderr = String::new(); - if let Some(mut pipe) = child.stderr.take() { - let _ = pipe.read_to_string(&mut stderr); - } - warn!( - "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed (exit {status}): {stderr}" - ); - return; - } - Ok(None) if Instant::now() >= deadline => { - let _ = child.kill(); - let _ = child.wait(); - warn!( - "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} timed out after {DOCKER_RM_TIMEOUT_SECS}s" - ); - return; - } - Ok(None) => std::thread::sleep(Duration::from_millis(100)), - Err(error) => { - warn!("Failed waiting for docker rm '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); - return; - } + Err(_) => { + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} timed out after {DOCKER_RM_TIMEOUT_SECS}s" + ); } } } diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs index 03313db377..872d26f8d6 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs @@ -33,6 +33,7 @@ use uuid::Uuid; const SINK_INDEX_PREFIX: &str = "iggy_messages"; const POLL_ATTEMPTS: usize = 100; const POLL_INTERVAL_MS: u64 = 50; +const REFRESH_PROBE_TIMEOUT_MS: u64 = 2_000; pub struct ElasticsearchSinkFixture { container: ElasticsearchContainer, @@ -64,8 +65,9 @@ impl ElasticsearchSinkFixture { let mut last_error: Option = None; for _ in 0..POLL_ATTEMPTS { - // Refresh so near-real-time search/count sees recently indexed docs. - if let Err(error) = self.refresh_index().await { + // Short-timeout probe: create_http_client() is 30s + 3 retries and + // would inflate failure-path wait_for_documents up to minutes. + if let Err(error) = self.refresh_index_probe().await { last_error = Some(error); sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; continue; @@ -104,6 +106,30 @@ impl ElasticsearchSinkFixture { pub async fn refresh_index(&self) -> Result<(), TestBinaryError> { ElasticsearchOps::refresh_index(self, &self.index).await } + + /// Refresh with a short-timeout client for the document poll loop. + async fn refresh_index_probe(&self) -> Result<(), TestBinaryError> { + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(REFRESH_PROBE_TIMEOUT_MS)) + .build() + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to build refresh probe client: {error}"), + })?; + let url = format!("{}/{}/_refresh", self.container.base_url, self.index); + let response = client.post(&url).send().await.map_err(|error| { + TestBinaryError::InvalidState { + message: format!("Failed to refresh index: {error}"), + } + })?; + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(TestBinaryError::InvalidState { + message: format!("Failed to refresh index: status={status}, body={body}"), + }); + } + Ok(()) + } } #[async_trait] From d0edc3bbcffe89100ca87c01066d2086e6e6cef0 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Fri, 24 Jul 2026 13:53:51 -0400 Subject: [PATCH 7/8] Fixing formating error --- .../tests/connectors/fixtures/elasticsearch/sink.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs index 872d26f8d6..f8668ac254 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs @@ -116,11 +116,14 @@ impl ElasticsearchSinkFixture { message: format!("Failed to build refresh probe client: {error}"), })?; let url = format!("{}/{}/_refresh", self.container.base_url, self.index); - let response = client.post(&url).send().await.map_err(|error| { - TestBinaryError::InvalidState { - message: format!("Failed to refresh index: {error}"), - } - })?; + let response = + client + .post(&url) + .send() + .await + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to refresh index: {error}"), + })?; if !response.status().is_success() { let status = response.status(); let body = response.text().await.unwrap_or_default(); From dbf3ef794491d1427520504b64f15c2570795624 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Fri, 24 Jul 2026 19:16:57 -0400 Subject: [PATCH 8/8] To reTrigger CI --- core/connectors/sinks/elasticsearch_sink/README.md | 2 +- core/connectors/sinks/elasticsearch_sink/config.toml | 2 -- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/core/connectors/sinks/elasticsearch_sink/README.md b/core/connectors/sinks/elasticsearch_sink/README.md index 0ba651b941..0c9c1038f4 100644 --- a/core/connectors/sinks/elasticsearch_sink/README.md +++ b/core/connectors/sinks/elasticsearch_sink/README.md @@ -8,7 +8,7 @@ A sink connector that consumes messages from Iggy streams and indexes them to El - `index`: Target index name - `username/password`: Optional authentication credentials - `batch_size`: Bulk indexing batch size (default: 100) -- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed (see #2927/#2928) +- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed - `create_index_if_not_exists`: Automatically create index (default: true) - `index_mapping`: Index mapping configuration diff --git a/core/connectors/sinks/elasticsearch_sink/config.toml b/core/connectors/sinks/elasticsearch_sink/config.toml index 75c6ed4748..171e0eec60 100644 --- a/core/connectors/sinks/elasticsearch_sink/config.toml +++ b/core/connectors/sinks/elasticsearch_sink/config.toml @@ -36,6 +36,4 @@ url = "http://localhost:9200" index = "iggy_messages" create_index_if_not_exists = true batch_size = 100 -# Client-wide timeout for open() and bulk consume(). 0 is clamped to 1s. -# Raise for slow bulk; timed-out consume drops the batch after poll commit (#2927/#2928). timeout_seconds = 30