fix(connectors): Harden Elasticsearch sink test readiness

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
 - `create_index_if_not_exists`: Automatically create index (default: true)
 - `index_mapping`: Index mapping configuration
 

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,
@@ -51,6 +54,11 @@ pub struct ElasticsearchSinkConfig {
     #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")]
     pub password: Option<SecretString>,
     pub batch_size: Option<usize>,
+    /// 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<u64>,
     pub create_index_if_not_exists: Option<bool>,
     pub index_mapping: Option<serde_json::Value>,
@@ -83,7 +91,17 @@ 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. 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
+            .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 =

core/integration/src/harness/handle/connectors_runtime.rs

@@ -243,19 +243,41 @@ impl IggyServerDependent for ConnectorsRuntimeHandle {
     }
 
     async fn wait_ready(&mut self) -> Result<(), TestBinaryError> {
-        let http_address = self.http_url();
-        let client = reqwest::Client::new();
+        // 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());
+        // 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 {
-            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,
                         });
                     }