This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new a8dd3a9c9 test(connectors): share one Elasticsearch container across
tests (#3401)
a8dd3a9c9 is described below
commit a8dd3a9c918e7563c6d18da1d351742aa5ef8aea
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Mon Jun 1 14:42:07 2026 +0200
test(connectors): share one Elasticsearch container across tests (#3401)
---
.claude/skills/connector-testing/SKILL.md | 17 +++++++++++++++
.config/nextest.toml | 15 ++++++++++++--
.../tests/connectors/fixtures/delta/fixture.rs | 5 +++--
.../tests/connectors/fixtures/doris/container.rs | 10 ++++-----
.../connectors/fixtures/elasticsearch/container.rs | 20 ++++++++++++------
.../connectors/fixtures/elasticsearch/sink.rs | 20 +++++++++++-------
.../connectors/fixtures/elasticsearch/source.rs | 20 +++++++++++-------
.../tests/connectors/fixtures/http/container.rs | 4 +++-
.../tests/connectors/fixtures/iceberg/container.rs | 8 +++++---
.../connectors/fixtures/influxdb/container.rs | 4 +++-
.../connectors/fixtures/influxdb/container_v3.rs | 2 ++
core/integration/tests/connectors/fixtures/mod.rs | 18 ++++++++++++++++
.../tests/connectors/fixtures/mongodb/container.rs | 5 ++++-
.../connectors/fixtures/postgres/container.rs | 14 ++++++++-----
.../connectors/fixtures/quickwit/container.rs | 10 +++++----
.../tests/connectors/fixtures/wiremock.rs | 4 +++-
justfile | 24 ++++++++++++++++++++++
17 files changed, 155 insertions(+), 45 deletions(-)
diff --git a/.claude/skills/connector-testing/SKILL.md
b/.claude/skills/connector-testing/SKILL.md
index ee56ed45c..5aa337907 100644
--- a/.claude/skills/connector-testing/SKILL.md
+++ b/.claude/skills/connector-testing/SKILL.md
@@ -184,6 +184,23 @@ core/integration/tests/connectors/
└── api/ HTTP control plane tests
```
+### Container naming + cleanup
+
+Every test container MUST be named with the `iggy-test-` prefix so a single
+`docker ps -aqf 'name=^iggy-test-'` sweep reaps them (`just
clean-test-containers`,
+alias `ctc`; the `test`/`nextest` recipes also reap on exit). New fixtures get
+this for free by naming through `fixtures::unique_container_name("<svc>")`,
which
+appends a uuid for parallel-safe ephemeral containers. Reuse fixtures use a
fixed
+`iggy-test-<svc>` literal instead, because the stable name is what lets later
+nextest processes attach to the same container via `ReuseDirective::Always`. Do
+not hand-roll a different prefix - the sweep only finds `iggy-test-*`.
+
+As of now only `elasticsearch` and `doris` share a container. Both have a slow
+boot (JVM / Doris FE+BE, tens of seconds), so reusing one across the whole
+regression beats spinning a fresh one per test. Every other fixture boots a
+fresh, ephemeral container per test - cheap enough that the isolation is worth
+more than the reuse.
+
### The `#[iggy_harness]` proc macro
Each integration test is annotated with this macro from the `integration`
crate. It boots an in-process Iggy server (and optionally the connectors
runtime) for the duration of the test, runs the seeds, then injects a
`&TestHarness` and your fixture.
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 0f4f9a2c3..3070df8bd 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -35,10 +35,21 @@ max-threads = 1
[[profile.default.overrides]]
filter = 'package(integration) and test(/connectors::doris::/)'
test-group = "doris"
-slow-timeout = { period = "60s", terminate-after = 8 }
+
+# Elasticsearch tests share one reusable container (fixed name
+# `iggy-test-elasticsearch`, ReuseDirective::Always). Serializing the group
+# lets the first test create it and the rest attach by name, instead of racing
+# to create the same name concurrently (Docker 409 Conflict). Per-test
+# isolation comes from a unique index per fixture, not a fresh container.
+[test-groups.elasticsearch]
+max-threads = 1
+
+[[profile.default.overrides]]
+filter = 'package(integration) and test(/connectors::elasticsearch::/)'
+test-group = "elasticsearch"
[profile.default]
-slow-timeout = { period = "30s", terminate-after = 4 }
+slow-timeout = { period = "60s", terminate-after = 5 }
[profile.ci]
retries = 3
diff --git a/core/integration/tests/connectors/fixtures/delta/fixture.rs
b/core/integration/tests/connectors/fixtures/delta/fixture.rs
index 0187e4465..6f9190a16 100644
--- a/core/integration/tests/connectors/fixtures/delta/fixture.rs
+++ b/core/integration/tests/connectors/fixtures/delta/fixture.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use async_trait::async_trait;
use deltalake::kernel::{DataType, PrimitiveType, StructField};
use deltalake::operations::create::CreateBuilder;
@@ -44,7 +45,7 @@ const ENV_SINK_AWS_S3_ENDPOINT_URL: &str =
const ENV_SINK_AWS_S3_ALLOW_HTTP: &str =
"IGGY_CONNECTORS_SINK_DELTA_PLUGIN_CONFIG_AWS_S3_ALLOW_HTTP";
-const MINIO_IMAGE: &str = "minio/minio";
+const MINIO_IMAGE: &str = "docker.io/minio/minio";
const MINIO_TAG: &str = "RELEASE.2025-09-07T16-13-09Z";
const MINIO_PORT: u16 = 9000;
const MINIO_CONSOLE_PORT: u16 = 9001;
@@ -348,7 +349,7 @@ impl TestFixture for DeltaS3Fixture {
async fn setup() -> Result<Self, TestBinaryError> {
let id = Uuid::new_v4();
let network = format!("iggy-delta-s3-{id}");
- let minio_name = format!("minio-delta-{id}");
+ let minio_name = fixtures::unique_container_name("minio-delta");
let (minio, minio_endpoint) = Self::start_minio(&network,
&minio_name).await?;
Self::create_bucket(&minio_endpoint).await?;
diff --git a/core/integration/tests/connectors/fixtures/doris/container.rs
b/core/integration/tests/connectors/fixtures/doris/container.rs
index e18e53a60..0a1fbe141 100644
--- a/core/integration/tests/connectors/fixtures/doris/container.rs
+++ b/core/integration/tests/connectors/fixtures/doris/container.rs
@@ -19,7 +19,7 @@
// Doris fixture. One Doris container is shared across every doris test in this
// CI job (or local `cargo test` session) via testcontainers'
reusable-containers
-// feature: the container is named `iggy-doris-test` and marked
+// feature: the container is named `iggy-test-doris` and marked
// `ReuseDirective::Always`, so the first test creates it and every subsequent
// test (in this or any other test process on the same Docker daemon) attaches
// to it. This is what makes `cargo test` and CI follow the same path — the old
@@ -31,7 +31,7 @@
//
// The container outlives the test session by design (that's what enables
// reuse). CI runners are ephemeral so it dies with them; locally, `docker rm
-// -f iggy-doris-test` forces a fresh boot.
+// -f iggy-test-doris` forces a fresh boot.
//
// Two host-level prerequisites cannot live inside the container:
// * `vm.max_map_count >= 2_000_000` — Doris 4.0.3's `start_be.sh` hard-exits
@@ -57,7 +57,7 @@ use tokio::time::sleep;
use tracing::info;
use uuid::Uuid;
-const DORIS_IMAGE: &str = "apache/doris";
+const DORIS_IMAGE: &str = "docker.io/apache/doris";
// Apache's maintained single-container line is now tagged `<version>-all` /
// `<version>-all-slim` (the old one-off `doris-all-in-one-2.1.0` was pushed
once
// in 2024 and never refreshed). `-slim` is the smaller base, so it pulls
faster
@@ -66,7 +66,7 @@ const DORIS_TAG: &str = "4.0.3-all-slim";
// Fixed name + `ReuseDirective::Always` is what makes the container survive
// across nextest's per-test processes. Stable name means every test process
// inspecting the Docker daemon finds the same one.
-const DORIS_CONTAINER_NAME: &str = "iggy-doris-test";
+const DORIS_CONTAINER_NAME: &str = "iggy-test-doris";
const FE_HTTP_PORT: u16 = 8030;
const FE_MYSQL_PORT: u16 = 9030;
const BE_HTTP_PORT: u16 = 8040;
@@ -232,7 +232,7 @@ impl DorisContainer {
//
// `with_container_name` + `with_reuse(Always)` is what makes the
// container survive across nextest's per-test processes: the first
- // test creates `iggy-doris-test`, every later test (in any process)
+ // test creates `iggy-test-doris`, every later test (in any process)
// attaches to it. The 1:1 BE port is therefore held continuously by
// one container, never racing with itself across container restarts.
let container = GenericImage::new(DORIS_IMAGE, DORIS_TAG)
diff --git
a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs
b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs
index 3eacc3a0a..32d15f069 100644
--- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs
+++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs
@@ -25,14 +25,20 @@ use serde::Deserialize;
use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy;
use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
use testcontainers_modules::testcontainers::runners::AsyncRunner;
-use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage,
ImageExt};
+use testcontainers_modules::testcontainers::{
+ ContainerAsync, GenericImage, ImageExt, ReuseDirective,
+};
use tracing::info;
-use uuid::Uuid;
-const ELASTICSEARCH_IMAGE: &str = "elasticsearch";
+const ELASTICSEARCH_IMAGE: &str = "docker.io/library/elasticsearch";
const ELASTICSEARCH_TAG: &str = "9.3.0";
const ELASTICSEARCH_PORT: u16 = 9200;
const ELASTICSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health";
+// Fixed name + ReuseDirective::Always shares one container across nextest's
+// per-test processes: the first test creates it, every later test attaches by
+// name. Per-test isolation comes from a unique index per fixture, not a fresh
+// container.
+const ELASTICSEARCH_CONTAINER_NAME: &str = "iggy-test-elasticsearch";
pub const DEFAULT_TEST_STREAM: &str = "test_stream";
pub const DEFAULT_TEST_TOPIC: &str = "test_topic";
@@ -88,6 +94,8 @@ pub struct ElasticsearchHit {
}
pub struct ElasticsearchContainer {
+ // Held so testcontainers' Drop runs on test exit; ReuseDirective::Always
+ // makes that Drop leave the container running for the next test to attach.
#[allow(dead_code)]
container: ContainerAsync<GenericImage>,
pub base_url: String,
@@ -95,8 +103,6 @@ pub struct ElasticsearchContainer {
impl ElasticsearchContainer {
pub async fn start() -> Result<Self, TestBinaryError> {
- let unique_network = format!("iggy-elasticsearch-source-{}",
Uuid::new_v4());
-
let container = GenericImage::new(ELASTICSEARCH_IMAGE,
ELASTICSEARCH_TAG)
.with_exposed_port(ELASTICSEARCH_PORT.tcp())
.with_wait_for(WaitFor::http(
@@ -104,11 +110,13 @@ impl ElasticsearchContainer {
.with_port(ELASTICSEARCH_PORT.tcp())
.with_expected_status_code(200u16),
))
- .with_network(unique_network)
+ .with_startup_timeout(std::time::Duration::from_secs(120))
.with_env_var("discovery.type", "single-node")
.with_env_var("xpack.security.enabled", "false")
.with_env_var("ES_JAVA_OPTS", "-Xms512m -Xmx512m")
.with_mapped_port(0, ELASTICSEARCH_PORT.tcp())
+ .with_container_name(ELASTICSEARCH_CONTAINER_NAME)
+ .with_reuse(ReuseDirective::Always)
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs
b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs
index 334dc72f1..bce21d5ea 100644
--- a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs
+++ b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs
@@ -30,14 +30,18 @@ use std::collections::HashMap;
use std::time::Duration;
use tokio::time::sleep;
use tracing::info;
+use uuid::Uuid;
-const SINK_INDEX: &str = "iggy_messages";
+const SINK_INDEX_PREFIX: &str = "iggy_messages";
const POLL_ATTEMPTS: usize = 100;
const POLL_INTERVAL_MS: u64 = 50;
pub struct ElasticsearchSinkFixture {
container: ElasticsearchContainer,
http_client: HttpClient,
+ // Unique per fixture so tests sharing one container never collide on the
+ // same index. The connector writes here via ENV_SINK_INDEX.
+ index: String,
}
impl ElasticsearchOps for ElasticsearchSinkFixture {
@@ -52,7 +56,7 @@ impl ElasticsearchOps for ElasticsearchSinkFixture {
impl ElasticsearchSinkFixture {
pub async fn get_document_count(&self) -> Result<usize, TestBinaryError> {
- self.count_documents(SINK_INDEX).await
+ self.count_documents(&self.index).await
}
pub async fn wait_for_documents(
@@ -60,7 +64,7 @@ impl ElasticsearchSinkFixture {
expected_count: usize,
) -> Result<usize, TestBinaryError> {
for _ in 0..POLL_ATTEMPTS {
- match self.count_documents(SINK_INDEX).await {
+ 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);
@@ -71,7 +75,7 @@ impl ElasticsearchSinkFixture {
sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
}
- let final_count = self.count_documents(SINK_INDEX).await.unwrap_or(0);
+ let final_count = self.count_documents(&self.index).await.unwrap_or(0);
Err(TestBinaryError::InvalidState {
message: format!(
"Expected at least {expected_count} documents, found
{final_count} after {POLL_ATTEMPTS} attempts"
@@ -80,11 +84,11 @@ impl ElasticsearchSinkFixture {
}
pub async fn search_documents(&self) ->
Result<ElasticsearchSearchResponse, TestBinaryError> {
- self.search_all(SINK_INDEX).await
+ self.search_all(&self.index).await
}
pub async fn refresh_index(&self) -> Result<(), TestBinaryError> {
- ElasticsearchOps::refresh_index(self, SINK_INDEX).await
+ ElasticsearchOps::refresh_index(self, &self.index).await
}
}
@@ -93,19 +97,21 @@ impl TestFixture for ElasticsearchSinkFixture {
async fn setup() -> Result<Self, TestBinaryError> {
let container = ElasticsearchContainer::start().await?;
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,
+ index,
})
}
fn connectors_runtime_envs(&self) -> HashMap<String, String> {
let mut envs = HashMap::new();
envs.insert(ENV_SINK_URL.to_string(), self.container.base_url.clone());
- envs.insert(ENV_SINK_INDEX.to_string(), SINK_INDEX.to_string());
+ envs.insert(ENV_SINK_INDEX.to_string(), self.index.clone());
envs.insert(
ENV_SINK_STREAMS_0_STREAM.to_string(),
DEFAULT_TEST_STREAM.to_string(),
diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs
b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs
index 16bdaaeae..8f8bb4a1a 100644
--- a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs
+++ b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs
@@ -28,13 +28,17 @@ use iggy_common::IggyTimestamp;
use integration::harness::{TestBinaryError, TestFixture};
use reqwest_middleware::ClientWithMiddleware as HttpClient;
use std::collections::HashMap;
+use uuid::Uuid;
-const TEST_INDEX: &str = "test_documents";
+const TEST_INDEX_PREFIX: &str = "test_documents";
/// Elasticsearch source fixture for basic document polling.
pub struct ElasticsearchSourceFixture {
container: ElasticsearchContainer,
http_client: HttpClient,
+ // Unique per fixture so tests sharing one container never collide on the
+ // same index. The connector reads from here via ENV_SOURCE_INDEX.
+ index: String,
}
impl ElasticsearchOps for ElasticsearchSourceFixture {
@@ -50,11 +54,11 @@ impl ElasticsearchOps for ElasticsearchSourceFixture {
impl ElasticsearchSourceFixture {
#[allow(dead_code)]
pub fn index_name(&self) -> &str {
- TEST_INDEX
+ &self.index
}
pub async fn setup_index(&self) -> Result<(), TestBinaryError> {
- self.create_index(TEST_INDEX).await
+ self.create_index(&self.index).await
}
pub async fn insert_document(
@@ -70,7 +74,7 @@ impl ElasticsearchSourceFixture {
"value": value,
"timestamp": timestamp
});
- self.index_document(TEST_INDEX, &doc_id.to_string(), &document)
+ self.index_document(&self.index, &doc_id.to_string(), &document)
.await
}
@@ -84,11 +88,11 @@ impl ElasticsearchSourceFixture {
}
pub async fn get_document_count(&self) -> Result<usize, TestBinaryError> {
- self.count_documents(TEST_INDEX).await
+ self.count_documents(&self.index).await
}
pub async fn refresh_index(&self) -> Result<(), TestBinaryError> {
- ElasticsearchOps::refresh_index(self, TEST_INDEX).await
+ ElasticsearchOps::refresh_index(self, &self.index).await
}
}
@@ -97,19 +101,21 @@ impl TestFixture for ElasticsearchSourceFixture {
async fn setup() -> Result<Self, TestBinaryError> {
let container = ElasticsearchContainer::start().await?;
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,
+ index,
})
}
fn connectors_runtime_envs(&self) -> HashMap<String, String> {
let mut envs = HashMap::new();
envs.insert(ENV_SOURCE_URL.to_string(),
self.container.base_url.clone());
- envs.insert(ENV_SOURCE_INDEX.to_string(), TEST_INDEX.to_string());
+ envs.insert(ENV_SOURCE_INDEX.to_string(), self.index.clone());
envs.insert(ENV_SOURCE_POLLING_INTERVAL.to_string(),
"100ms".to_string());
envs.insert(ENV_SOURCE_BATCH_SIZE.to_string(), "100".to_string());
envs.insert(
diff --git a/core/integration/tests/connectors/fixtures/http/container.rs
b/core/integration/tests/connectors/fixtures/http/container.rs
index 1f5204871..669166aff 100644
--- a/core/integration/tests/connectors/fixtures/http/container.rs
+++ b/core/integration/tests/connectors/fixtures/http/container.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use integration::harness::TestBinaryError;
use std::time::Duration;
use testcontainers_modules::testcontainers::core::WaitFor::Healthcheck;
@@ -27,7 +28,7 @@ use testcontainers_modules::testcontainers::{ContainerAsync,
GenericImage, Image
use tokio::time::sleep;
use tracing::info;
-const WIREMOCK_IMAGE: &str = "wiremock/wiremock";
+const WIREMOCK_IMAGE: &str = "docker.io/wiremock/wiremock";
const WIREMOCK_TAG: &str = "3.13.2";
const WIREMOCK_PORT: u16 = 8080;
@@ -86,6 +87,7 @@ impl HttpSinkWireMockContainer {
.to_string(),
"/home/wiremock/mappings",
))
+
.with_container_name(fixtures::unique_container_name("wiremock-http"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
diff --git a/core/integration/tests/connectors/fixtures/iceberg/container.rs
b/core/integration/tests/connectors/fixtures/iceberg/container.rs
index c0bcc5c13..c4ad10f40 100644
--- a/core/integration/tests/connectors/fixtures/iceberg/container.rs
+++ b/core/integration/tests/connectors/fixtures/iceberg/container.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use async_trait::async_trait;
use integration::harness::{TestBinaryError, TestFixture};
use reqwest_middleware::ClientWithMiddleware as HttpClient;
@@ -30,11 +31,11 @@ use
testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, Image
use tracing::info;
use uuid::Uuid;
-const MINIO_IMAGE: &str = "minio/minio";
+const MINIO_IMAGE: &str = "docker.io/minio/minio";
const MINIO_TAG: &str = "RELEASE.2025-09-07T16-13-09Z";
const MINIO_PORT: u16 = 9000;
const MINIO_CONSOLE_PORT: u16 = 9001;
-const ICEBERG_REST_IMAGE: &str = "apache/iceberg-rest-fixture";
+const ICEBERG_REST_IMAGE: &str = "docker.io/apache/iceberg-rest-fixture";
const ICEBERG_REST_TAG: &str = "latest";
const ICEBERG_REST_PORT: u16 = 8181;
@@ -150,6 +151,7 @@ impl IcebergRestContainer {
.with_env_var("AWS_ACCESS_KEY_ID", MINIO_ACCESS_KEY)
.with_env_var("AWS_SECRET_ACCESS_KEY", MINIO_SECRET_KEY)
.with_mapped_port(0, ICEBERG_REST_PORT.tcp())
+
.with_container_name(fixtures::unique_container_name("iceberg-rest"))
.start()
.await
.map_err(|error| TestBinaryError::FixtureSetup {
@@ -448,7 +450,7 @@ impl TestFixture for IcebergFixture {
async fn setup() -> Result<Self, TestBinaryError> {
let id = Uuid::new_v4();
let network = format!("iggy-iceberg-{id}");
- let minio_name = format!("minio-{id}");
+ let minio_name = fixtures::unique_container_name("minio-iceberg");
let minio = MinioContainer::start(&network, &minio_name).await?;
diff --git a/core/integration/tests/connectors/fixtures/influxdb/container.rs
b/core/integration/tests/connectors/fixtures/influxdb/container.rs
index dcf707c52..914430bb3 100644
--- a/core/integration/tests/connectors/fixtures/influxdb/container.rs
+++ b/core/integration/tests/connectors/fixtures/influxdb/container.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use integration::harness::TestBinaryError;
use reqwest_middleware::ClientWithMiddleware as HttpClient;
use reqwest_retry::RetryTransientMiddleware;
@@ -27,7 +28,7 @@ use
testcontainers_modules::testcontainers::runners::AsyncRunner;
use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage,
ImageExt};
use tracing::info;
-const INFLUXDB_IMAGE: &str = "influxdb";
+const INFLUXDB_IMAGE: &str = "docker.io/library/influxdb";
const INFLUXDB_TAG: &str = "2.7-alpine";
const INFLUXDB_PORT: u16 = 8086;
@@ -117,6 +118,7 @@ impl InfluxDbContainer {
.with_env_var("DOCKER_INFLUXDB_INIT_ORG", INFLUXDB_ORG)
.with_env_var("DOCKER_INFLUXDB_INIT_BUCKET", INFLUXDB_BUCKET)
.with_env_var("DOCKER_INFLUXDB_INIT_ADMIN_TOKEN",
INFLUXDB_TOKEN)
+
.with_container_name(fixtures::unique_container_name("influxdb"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
diff --git
a/core/integration/tests/connectors/fixtures/influxdb/container_v3.rs
b/core/integration/tests/connectors/fixtures/influxdb/container_v3.rs
index 138ccdef9..dcbe63402 100644
--- a/core/integration/tests/connectors/fixtures/influxdb/container_v3.rs
+++ b/core/integration/tests/connectors/fixtures/influxdb/container_v3.rs
@@ -26,6 +26,7 @@
//! - Query: `POST /api/v3/query_sql` with `{"db":…,"q":…,"format":"jsonl"}`
//! - Auth: `Authorization: Bearer {token}`
+use crate::connectors::fixtures;
use integration::harness::TestBinaryError;
use reqwest_middleware::ClientWithMiddleware as HttpClient;
use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
@@ -115,6 +116,7 @@ impl InfluxDb3Container {
"memory",
"--without-auth",
])
+
.with_container_name(fixtures::unique_container_name("influxdb3"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
diff --git a/core/integration/tests/connectors/fixtures/mod.rs
b/core/integration/tests/connectors/fixtures/mod.rs
index 2951cacbe..616c3a557 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -17,6 +17,8 @@
* under the License.
*/
+use uuid::Uuid;
+
mod delta;
mod doris;
mod elasticsearch;
@@ -28,6 +30,22 @@ mod postgres;
mod quickwit;
mod wiremock;
+/// Prefix on every test container name so `just clean-test-containers` reaps
+/// them all with one `name=^iggy-test-` filter. A new fixture only has to use
+/// `unique_container_name` (or a fixed `iggy-test-<svc>` for reuse containers)
+/// to be covered.
+pub(crate) const TEST_CONTAINER_PREFIX: &str = "iggy-test-";
+
+/// Unique per-test container name for ephemeral fixtures. Reuse fixtures
+/// (elasticsearch, doris) use a fixed `iggy-test-<svc>` literal instead, since
+/// the stable name is what lets later test processes attach to the same one.
+pub(crate) fn unique_container_name(service: &str) -> String {
+ format!(
+ "{TEST_CONTAINER_PREFIX}{service}-{}",
+ Uuid::new_v4().simple()
+ )
+}
+
pub use delta::{DeltaFixture, DeltaS3Fixture};
pub use doris::{
DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture,
DorisSinkMaxFilterRatioFixture,
diff --git a/core/integration/tests/connectors/fixtures/mongodb/container.rs
b/core/integration/tests/connectors/fixtures/mongodb/container.rs
index 4c2b5177d..f58bd3e27 100644
--- a/core/integration/tests/connectors/fixtures/mongodb/container.rs
+++ b/core/integration/tests/connectors/fixtures/mongodb/container.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use integration::harness::TestBinaryError;
use mongodb::{Client, bson::doc, options::ClientOptions};
use std::time::Duration;
@@ -26,7 +27,7 @@ use testcontainers_modules::testcontainers::{ContainerAsync,
GenericImage, Image
use tokio::time::sleep;
use tracing::info;
-const MONGODB_IMAGE: &str = "mongo";
+const MONGODB_IMAGE: &str = "docker.io/library/mongo";
const MONGODB_TAG: &str = "7";
const MONGODB_PORT: u16 = 27017;
const MONGODB_READY_MSG: &str = "Waiting for connections";
@@ -80,6 +81,7 @@ impl MongoDbContainer {
.with_exposed_port(MONGODB_PORT.tcp())
.with_wait_for(WaitFor::message_on_stdout(MONGODB_READY_MSG))
.with_mapped_port(0, MONGODB_PORT.tcp())
+ .with_container_name(fixtures::unique_container_name("mongodb"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
@@ -121,6 +123,7 @@ impl MongoDbContainer {
.with_exposed_port(MONGODB_PORT.tcp())
.with_wait_for(WaitFor::message_on_stdout(MONGODB_READY_MSG))
.with_mapped_port(0, MONGODB_PORT.tcp())
+ .with_container_name(fixtures::unique_container_name("mongodb-rs"))
.with_cmd(["--replSet", MONGODB_REPLICA_SET_NAME,
"--bind_ip_all"]);
if enable_test_commands {
diff --git a/core/integration/tests/connectors/fixtures/postgres/container.rs
b/core/integration/tests/connectors/fixtures/postgres/container.rs
index 51478810f..b2c5ae288 100644
--- a/core/integration/tests/connectors/fixtures/postgres/container.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/container.rs
@@ -20,9 +20,11 @@
use integration::harness::TestBinaryError;
use sqlx::postgres::PgPoolOptions;
use sqlx::{Pool, Postgres};
+
+use crate::connectors::fixtures;
use testcontainers_modules::{
postgres,
- testcontainers::{ContainerAsync, runners::AsyncRunner},
+ testcontainers::{ContainerAsync, ImageExt, runners::AsyncRunner},
};
pub(super) const POSTGRES_PORT: u16 = 5432;
@@ -128,12 +130,14 @@ pub struct PostgresContainer {
impl PostgresContainer {
pub(super) async fn start() -> Result<Self, TestBinaryError> {
- let container =
postgres::Postgres::default().start().await.map_err(|e| {
- TestBinaryError::FixtureSetup {
+ let container = postgres::Postgres::default()
+ .with_container_name(fixtures::unique_container_name("postgres"))
+ .start()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
fixture_type: "PostgresContainer".to_string(),
message: format!("Failed to start container: {e}"),
- }
- })?;
+ })?;
let host_port = container
.get_host_port_ipv4(POSTGRES_PORT)
diff --git a/core/integration/tests/connectors/fixtures/quickwit/container.rs
b/core/integration/tests/connectors/fixtures/quickwit/container.rs
index 9d062937c..de0760b83 100644
--- a/core/integration/tests/connectors/fixtures/quickwit/container.rs
+++ b/core/integration/tests/connectors/fixtures/quickwit/container.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use async_trait::async_trait;
use integration::harness::{TestBinaryError, TestFixture, seeds};
use reqwest_middleware::ClientWithMiddleware as HttpClient;
@@ -33,10 +34,10 @@ use tokio::time::sleep;
use tracing::info;
use uuid::Uuid;
-const DEFAULT_POLL_ATTEMPTS: usize = 100;
-const DEFAULT_POLL_INTERVAL_MS: u64 = 50;
+const DEFAULT_POLL_ATTEMPTS: usize = 600;
+const DEFAULT_POLL_INTERVAL_MS: u64 = 100;
-const QUICKWIT_IMAGE: &str = "quickwit/quickwit";
+const QUICKWIT_IMAGE: &str = "docker.io/quickwit/quickwit";
const QUICKWIT_TAG: &str = "0.8.2";
const QUICKWIT_PORT: u16 = 7280;
const QUICKWIT_LISTEN_ADDRESS: &str = "0.0.0.0";
@@ -78,6 +79,7 @@ impl QuickwitContainer {
.with_cmd(["run"])
.with_env_var("QW_LISTEN_ADDRESS", QUICKWIT_LISTEN_ADDRESS)
.with_mapped_port(0, QUICKWIT_PORT.tcp())
+ .with_container_name(fixtures::unique_container_name("quickwit"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
@@ -253,7 +255,7 @@ pub trait QuickwitOps: Sync {
return Ok(search);
}
}
- sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS /
5)).await;
+ sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS)).await;
}
Err(TestBinaryError::InvalidState {
message: format!(
diff --git a/core/integration/tests/connectors/fixtures/wiremock.rs
b/core/integration/tests/connectors/fixtures/wiremock.rs
index ceda81949..75e886e97 100644
--- a/core/integration/tests/connectors/fixtures/wiremock.rs
+++ b/core/integration/tests/connectors/fixtures/wiremock.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use crate::connectors::fixtures;
use async_trait::async_trait;
use integration::harness::{TestBinaryError, TestFixture};
use std::collections::HashMap;
@@ -26,7 +27,7 @@ use
testcontainers_modules::testcontainers::core::{IntoContainerPort, Mount};
use testcontainers_modules::testcontainers::runners::AsyncRunner;
use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage,
ImageExt};
-const WIREMOCK_IMAGE: &str = "wiremock/wiremock";
+const WIREMOCK_IMAGE: &str = "docker.io/wiremock/wiremock";
const WIREMOCK_TAG: &str = "3.13.2";
const WIREMOCK_PORT: u16 = 8080;
@@ -62,6 +63,7 @@ impl WireMockContainer {
.to_string(),
"/home/wiremock/__files",
))
+ .with_container_name(fixtures::unique_container_name("wiremock"))
.start()
.await
.map_err(|e| TestBinaryError::FixtureSetup {
diff --git a/justfile b/justfile
index 7b68d0c4d..3ad75a6fd 100644
--- a/justfile
+++ b/justfile
@@ -32,22 +32,46 @@ alias pcs := profile-cpu-server
alias pcc := profile-cpu-client
alias pis := profile-io-server
alias pic := profile-io-client
+alias ctc := clean-test-containers
+
+# Every test container is named `iggy-test-<service>[-<uuid>]` (see
+# fixtures::unique_container_name). Reuse fixtures (elasticsearch, doris) keep
+# their container running, and a crashed run can orphan any fixture's
container,
+# so the recipes below reap everything matching the prefix on exit.
+reap_test_containers := "docker ps -aqf 'name=^iggy-test-' | xargs -r docker
rm -f >/dev/null 2>&1 || true"
build:
cargo build
test: build
+ #!/usr/bin/env bash
+ set -euo pipefail
+ trap "{{reap_test_containers}}" EXIT
cargo test
tests TEST: build
+ #!/usr/bin/env bash
+ set -euo pipefail
+ trap "{{reap_test_containers}}" EXIT
cargo test {{TEST}}
nextest: build
+ #!/usr/bin/env bash
+ set -euo pipefail
+ trap "{{reap_test_containers}}" EXIT
cargo nextest run
nextests TEST: build
+ #!/usr/bin/env bash
+ set -euo pipefail
+ trap "{{reap_test_containers}}" EXIT
cargo nextest run --nocapture -- {{TEST}}
+# Force-remove all `iggy-test-*` containers on demand. Useful after running
+# `cargo nextest run` directly (outside the recipes above).
+clean-test-containers:
+ {{reap_test_containers}}
+
# Run Miri (UB detector) on the unsafe-heavy crates that don't pull
# tokio/compio. Mirrors the `miri` task in CI. Pinned to the same nightly
# as `.github/actions/rust/pre-merge/action.yml` so local runs don't drift
