ryerraguntla commented on code in PR #3729:
URL: https://github.com/apache/iggy/pull/3729#discussion_r3831725349
##########
core/integration/tests/connectors/fixtures/elasticsearch/container.rs:
##########
@@ -37,6 +40,25 @@ 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";
+// 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;
Review Comment:
Fixed
##########
core/integration/tests/connectors/fixtures/elasticsearch/sink.rs:
##########
@@ -88,6 +106,74 @@ 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 = probe_client()?;
+ 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(())
+ }
+
+ /// Count with a short-timeout client for the document poll loop.
+ ///
+ /// `count_documents` (via `ElasticsearchOps`) goes through the shared
+ /// `create_http_client()` - 30s timeout plus 3 retries - so a degraded
+ /// `_count` endpoint could block this poll for minutes even though
+ /// `refresh_index_probe` already uses a short timeout.
+ async fn count_documents_probe(&self) -> Result<usize, TestBinaryError> {
+ let client = probe_client()?;
+ let url = format!("{}/{}/_count", self.container.base_url, self.index);
+ let response =
+ client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|error| TestBinaryError::InvalidState {
+ message: format!("Failed to count documents: {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 count documents: status={status},
body={body}"),
+ });
+ }
+
+ #[derive(serde::Deserialize)]
+ struct CountResponse {
+ count: usize,
+ }
+ let count_response =
response.json::<CountResponse>().await.map_err(|error| {
+ TestBinaryError::InvalidState {
+ message: format!("Failed to parse count response: {error}"),
+ }
+ })?;
+ Ok(count_response.count)
+ }
+}
+
+fn probe_client() -> Result<reqwest::Client, TestBinaryError> {
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
