This is an automated email from the ASF dual-hosted git repository.
chl-wxp pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new beae69d9d0 [Feature][Doc] Develop skills and coding guidelines for the
Test suite (#11212)
beae69d9d0 is described below
commit beae69d9d0588154895a5d453c50d08791bd2879
Author: zhiwei.niu <[email protected]>
AuthorDate: Thu Jul 2 09:16:50 2026 +0800
[Feature][Doc] Develop skills and coding guidelines for the Test suite
(#11212)
Co-authored-by: xuepeng <[email protected]>
---
.skills/seatunnel-test-suite/SKILL.md | 240 +++++++++++
.skills/seatunnel-test-suite/agents/openai.yaml | 19 +
docs/en/developer/test-coding-guide.md | 539 ++++++++++++++++++++++++
docs/sidebars.js | 1 +
docs/zh/developer/test-coding-guide.md | 518 +++++++++++++++++++++++
5 files changed, 1317 insertions(+)
diff --git a/.skills/seatunnel-test-suite/SKILL.md
b/.skills/seatunnel-test-suite/SKILL.md
new file mode 100644
index 0000000000..fc4c91bbd2
--- /dev/null
+++ b/.skills/seatunnel-test-suite/SKILL.md
@@ -0,0 +1,240 @@
+---
+name: seatunnel-test-suite
+description: Write or review Apache SeaTunnel E2E (Testcontainers) and unit
tests so they are stable, leak-free, and deterministic. Use when creating a new
*IT/*Test class, reviewing a test diff, or refactoring a flaky or leaking test.
+---
+
+# SeaTunnel Test Suite
+
+Self-contained operating manual for writing and reviewing SeaTunnel tests: the
rule tables, citable IDs, and
+the patterns needed to act. Everything required to apply or cite a rule is in
this file.
+
+## When to use
+
+- Writing a new connector E2E test (`*IT.java`) or unit test (`*Test.java`).
+- Reviewing a test diff for flakiness, leaks, or layout violations before a PR.
+- Stabilizing a flaky or leaking test, or bringing an older test up to these
conventions.
+
+## Framework contract (E2E)
+
+Every E2E test **must** follow this structural contract — violating it means
the test silently does nothing.
+
+| Requirement | Detail |
+|-------------|--------|
+| Extend `TestSuiteBase` | Provides the engine container lifecycle, `NETWORK`,
and test dispatch |
+| `@TestInstance(Lifecycle.PER_CLASS)` | Inherited from `TestSuiteBase` — one
instance shared across all methods; fields need not be `static` |
+| `@TestTemplate` on each test method | NOT `@Test`. The
`TestCaseInvocationContextProvider` only dispatches `@TestTemplate` methods |
+| `TestContainer` parameter | Each `@TestTemplate` method must accept a single
`TestContainer container` parameter |
+| `@DisabledOnContainer` (optional) | Add **only** when the scenario genuinely
can't run on an engine — see below. Default to no annotation so the test runs
on all engines. |
+
+Minimal skeleton — runs on every engine (Zeta, Flink, Spark), which is the
default and correct case for
+most batch source/sink tests:
+
+```java
+public class MyConnectorIT extends TestSuiteBase {
+
+ @TestTemplate
+ public void testSourceToSink(TestContainer container) throws Exception {
+ Container.ExecResult result =
container.executeJob("/my_connector_to_assert.conf");
+ Assertions.assertEquals(0, result.getExitCode());
+ }
+}
+```
+
+### When to add `@DisabledOnContainer`
+
+Do not exclude an engine by default. Add the annotation only when a scenario
provably cannot run on that
+engine, and word `disabledReason` for the *specific* scenario — not a blanket
"engine X is unsupported".
+The exclusion can go on the class (whole suite) or a single `@TestTemplate`
method (just that case). Real
+reasons from the codebase:
+
+| Excluded engines | Why (the real constraint) |
+|------------------|---------------------------|
+| Spark | CDC / streaming jobs — Spark doesn't support the
continuous/changelog mode |
+| Spark + Flink | checkpoint-restore tests; scenarios needing Zeta-only
features |
+| Spark + Flink (all but Zeta) | continuous-discovery long-running jobs;
SeaTunnel-only behavior |
+| Spark | drops the RowKind of a record, so changelog assertions fail |
+
+```java
+// Only this CDC case can't run on Spark — exclude at the method, not the
class:
+@TestTemplate
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK},
+ disabledReason = "Spark does not support the CDC streaming job")
+public void testCdcStreaming(TestContainer container) throws Exception {
+ // ...
+}
+```
+
+## The rules
+
+Two namespaces: **E1–E7 govern E2E (`*IT`) tests** (cite as `[E3]`); **U1–U6
govern unit (`*Test`)
+tests** (cite as `[U5]`). E6 is Zeta-engine-only. Pick the namespace that
matches the file under review —
+never apply an E-rule to a unit test or vice versa.
+
+| # | Rule | Check (the smell) | Fix |
+|---|------|-------------------|-----|
+| E1 | Dynamic ports | literal `127.0.0.1:<port>` / `localhost:<port>` in Java
code | `container.getHost()` + `container.getMappedPort(p)`; `.conf` files
reference the network alias + internal port (see E7) |
+| E2 | Condition-based waiting | any `Thread.sleep(...)` |
`Awaitility.await().atMost(...).pollInterval(...).untilAsserted(...)`; timeout
from the scenario table below |
+| E3 | Release resources | an opened client/connection/container with no close
| implement `TestResource`, close in `tearDown()` reverse-order + null-safe;
try-with-resources for method-scoped |
+| E4 | Async job submission | inline `executeJob` for a streaming/CDC job
(never returns) | `CompletableFuture.supplyAsync(...)`, gate on `RUNNING`, act,
verify, `cancel(true)` |
+| E5 | Share one container | a fresh container started per test method | share
one container across all methods in a class (`@TestInstance(PER_CLASS)`) |
+| E5b | Pin image versions | `:latest` image tag | pin a specific version,
e.g. `mysql:8.0.32`, never `mysql:latest` (see subsection) |
+| E5c | Cover all data types | only `String`/`int` exercised | cover the
connector's full data-type set, not just the easy ones |
+| E6 | Thread whitelist (Zeta) | `There are still threads running in the
container` | close the client first (E3); if the lib thread is unrecyclable,
whitelist its prefix in `isIssueWeAlreadyKnow(...)` |
+| E7 | Docker network | container not reachable from engine container |
`.withNetwork(NETWORK).withNetworkAliases("my-alias")`; start with
`Startables.deepStart(...)`; add
`Slf4jLogConsumer(DockerLoggerFactory.getLogger(IMAGE))` for debug logs (full
setup below) |
+
+### E7 — network setup
+
+The SeaTunnel engine runs **inside** its own Docker container. External
service containers must share the
+same Docker network so the engine can reach them (this is also why E1's
`.conf` files use the alias, not the
+mapped host port). Setup pattern:
+
+```java
+// In your IT class @BeforeAll (NETWORK is inherited from TestSuiteBase):
+container = new GenericContainer<>(DockerImageName.parse("my-service:1.2.3"))
+ .withNetwork(NETWORK) // inherited field
from TestSuiteBase
+ .withNetworkAliases("my-service-host") // reachable by this
name from engine
+ .withExposedPorts(3306)
+ .withLogConsumer(new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("my-service:1.2.3")));
+Startables.deepStart(Stream.of(container)).join(); // parallel startup
+```
+
+In your `.conf` file (runs inside the engine container):
+```hocon
+host = "my-service-host"
+port = 3306
+```
+
+In your Java test setup (runs on the host):
+```java
+String jdbcUrl = String.format("jdbc:mysql://%s:%d/test",
+ container.getHost(), container.getMappedPort(3306));
+```
+
+### E2 — timeout reference
+
+| Scenario | atMost | pollInterval |
+|----------|--------|--------------|
+| Container / client readiness | 2 min | 1 s |
+| Job reaches `RUNNING` | 1 min | 2 s |
+| Batch job result verified | 60 s | 2 s |
+| MQ / Kafka consumption | 30–60 s | 1 s |
+| CDC / schema-change propagation | 60–120 s | 2–5 s |
+
+Chain `.ignoreExceptions()` onto the `Awaitility.await()` (the E2 fix)
whenever the client throws until the
+service is up — otherwise the first poll's exception fails the wait instead of
retrying. Never pick a bare
+round number without a scenario to justify it.
+
+### E5b — image version pinning
+
+Never use `:latest` — upstream images can change behavior without warning,
breaking CI with no code diff to
+bisect. Pin to a specific minor or patch version:
+
+```java
+// Bad — breaks unpredictably
+private static final String IMAGE = "postgres:latest";
+
+// Good — reproducible
+private static final String IMAGE = "postgres:14.5";
+```
+
+### E6 — thread whitelist, in one paragraph
+
+After the last job finishes, `SeaTunnelContainer` snapshots the server JVM and
fails if a non-system thread
+survives 120s. Daemon threads from third-party clients (JDBC drivers, HTTP
pools) are sometimes unrecyclable.
+For those only, add a **specific name prefix** to
`isIssueWeAlreadyKnow(String)` in `SeaTunnelContainer`, with
+a comment naming the library. A thread you *can* close belongs in
`tearDown()`, not the whitelist. System
+threads (`hz.main`, `pool-N-thread-N`, …) are already handled by
`isSystemThread(...)` — don't duplicate them.
+
+### UT rules
+
+Use this table for `*Test` classes (Surefire, no engine container):
+
+| # | Rule | Smell | Fix |
+|---|------|-------|-----|
+| U1 | Test behavior, not internals | asserts on private fields or internal
call order | assert observable output/state; verify `void` calls only when
side-effect IS the contract |
+| U2 | No timing or IO dependence | `Thread.sleep`,
`System.currentTimeMillis()`, filesystem, network | remove timing; mock/fake IO
boundaries; inject clocks if needed |
+| U3 | Mock at boundaries only | mocking domain objects or `new`-able value
types | only mock external interfaces/clients/maps/services the class depends
on via constructor or setter |
+| U4 | One behavior per method | multi-branch `if`/`for` inside a single test
| split into one method per branch; name each method as the behavior it proves |
+| U5 | Negative paths assert message | `assertThrows(FooException.class, ...)`
with no message check | also call
`assertThat(ex.getMessage()).contains("expected fragment")` |
+| U6 | Naming | `testFoo`, `testBar`, `test123` | class `*Test`; methods like
`shouldReturn404WhenUserMissing`, `throwsOnNullHost` |
+
+**Mock setup shape** — each test method is Arrange-Act-Assert:
+1. *Arrange* — in `@BeforeEach`, `mock(...)` every external boundary, wire the
chain with `when(...).thenReturn(...)`, then construct the SUT. Per test, stub
only the values that select the branch under test.
+2. *Act* — call the one method under test.
+3. *Assert* — check observable output (U1), not call chains.
+
+Pitfalls that make a mocked test pass for the wrong reason:
+- When the SUT reaches a collaborator through a chain (`a.getB().getC(key)`),
stub the whole chain, and stub each lookup with the *exact* key constant the
production code uses — a mismatched key returns `null` and silently changes the
branch taken.
+- To exercise a downstream branch, stub every upstream guard non-null first; a
null upstream value short-circuits early, so stubbing only the downstream value
tests nothing.
+- For negative paths, assert the exception message fragment, not just the type
(U5) — a wrong-cause throw of the same type would otherwise pass.
+
+**When NOT to mock:**
+- Simple value objects, POJOs, DTOs — use real instances.
+- The class under test itself — never mock the SUT.
+- When an in-memory fake (e.g. `HashMap` instead of an interface) is simpler
and equally isolated.
+
+## Layout (where things go)
+
+One Maven module per connector under
`seatunnel-e2e/seatunnel-connector-v2-e2e/connector-<name>-e2e/`. Files
+load from the test **classpath root** (`src/test/resources/`):
+
+- Test class →
`src/test/java/org/apache/seatunnel/e2e/connector/<name>/<Name>IT.java`
+ (CDC connectors use the `...connectors.seatunnel.cdc.<name>` package). `*IT`
= integration (Failsafe),
+ `*Test` = unit (Surefire).
+- Job configs → `src/test/resources/<name>_source_to_sink.conf`, referenced
with a **leading slash**
+ (`executeJob("/<name>_source_to_sink.conf")`). License header required (`#`
comments).
+- DDL → `src/test/resources/ddl/<table>.sql` (CDC `UniqueDatabase` resolves
`ddl/<template>.sql`).
+- Container mounts → `src/test/resources/docker/...`, referenced via the
container API (no leading slash).
+- Register the module in the parent `pom.xml`; for a new connector also update
CI label and
+ `plugin-mapping.properties`.
+
+## Review workflow
+
+1. Grep for the obvious smells (a clean result is a hint, not a guarantee —
`127\.0\.0\.1` and `localhost`
+ are also matched without a port to catch host references where the port is
templated separately):
+ ```bash
+ grep -nE "Thread\.sleep|127\.0\.0\.1|localhost|:latest" <test-file>
+ ```
+2. **[E1, E7]** External containers joined to `NETWORK` with aliases? `.conf`
uses alias + internal port?
+3. **Framework contract:** extends `TestSuiteBase`? `@TestTemplate` +
`TestContainer` param (not `@Test`)?
+4. **`@DisabledOnContainer`** — present only where a scenario truly can't run
on an engine (no blanket exclusions), and scoped to the affected method when
only one case is incompatible?
+5. **[E3]** Each opened resource closed in `tearDown()` (or
try-with-resources)? Container stopped explicitly?
+6. **[E4]** Streaming/CDC jobs submitted via `CompletableFuture.supplyAsync`
and gated on `RUNNING`?
+7. **[E2]** Timeouts justified by the scenario table — no bare round numbers?
+8. **[E5, E5b, E5c]** One container shared per class? Image versions pinned
(no `:latest`)? Full type set?
+9. Resources in the conventional paths; new module registered in the parent
pom?
+10. **[E6]** For Zeta E2E, run the full job set; if the thread-leak check
fires, close the client first.
+11. For unit tests, walk the **[U1–U6]** table; for the negative path confirm
**[U5]** asserts the message.
+12. Run the targeted test class multiple times to confirm stability:
+ ```bash
+ ./mvnw -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-<name>-e2e \
+ -DskipUT -DskipIT=false -Dit.test=<ITClassName> verify
+ ```
+
+## Quick-fix reference
+
+Each row cites the rule it enforces, so a quick-fix finding maps directly to
the output format below.
+
+| Rule | Smell | Replace with |
+|------|-------|--------------|
+| E2 | `Thread.sleep(n)` then assert |
`Awaitility.await().atMost(...).untilAsserted(...)` |
+| E1 | `"tcp://localhost:61616"` | `container.getHost() + ":" +
container.getMappedPort(61616)` |
+| E3 | missing `tearDown()` | implement `TestResource`; null-safe,
reverse-order close + `container.stop()` |
+| E4 | inline `executeJob` for a streaming job |
`CompletableFuture.supplyAsync(...)` + gate on `RUNNING` |
+| E4 | `consumer.receive()` with no timeout | `consumer.receive(timeoutMs)` or
Awaitility poll |
+| E4 | swallowed exception in async block | rethrow as `CompletionException` |
+| E6 | Zeta E2E fails on leftover client thread | close in `tearDown()`; if
unrecyclable, whitelist in `isIssueWeAlreadyKnow` |
+| — | `@Test` on E2E method | change to `@TestTemplate` + add `TestContainer
container` parameter |
+| E5b | `postgres:latest` | `postgres:14.5` (pin to specific version) |
+| E7 | container unreachable from engine |
`.withNetwork(NETWORK).withNetworkAliases("...")` |
+| U5 | `assertThrows(...)` with no message check | add
`assertTrue(ex.getMessage().contains("..."))` |
+
+## Output format
+
+When reviewing, report each finding as `[<RuleId>] <file>:<line> — <issue>`
with a one-line fix, where
+`<RuleId>` is an E-rule (`E1`–`E7`) for `*IT` tests or a U-rule (`U1`–`U6`)
for `*Test` units — e.g.
+`[E3] FooIT.java:88 — Kafka consumer never closed`. Show a before/after
snippet only when the fix isn't
+obvious from the rule. When writing a new class, satisfy all applicable rules
and add the Apache License
+header to every new file (including `.conf` and `.sql`).
diff --git a/.skills/seatunnel-test-suite/agents/openai.yaml
b/.skills/seatunnel-test-suite/agents/openai.yaml
new file mode 100644
index 0000000000..a27d2dca18
--- /dev/null
+++ b/.skills/seatunnel-test-suite/agents/openai.yaml
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+interface:
+ display_name: "SeaTunnel Test Suite"
+ short_description: "Write and review high-quality SeaTunnel E2E & Unit tests"
+ default_prompt: "Use $seatunnel-test-suite to review or write a SeaTunnel
E2E test following stability best practices."
diff --git a/docs/en/developer/test-coding-guide.md
b/docs/en/developer/test-coding-guide.md
new file mode 100644
index 0000000000..357022d04e
--- /dev/null
+++ b/docs/en/developer/test-coding-guide.md
@@ -0,0 +1,539 @@
+---
+title: Test Coding Guide
+---
+
+# Test Coding Guide
+
+This guide describes how to write a high-quality, stable End-to-End (E2E) or
Unit Test for Apache SeaTunnel.
+It complements the [Coding Guide](coding-guide.md): the Coding Guide covers
general PR quality, while this
+guide focuses specifically on test stability and resource safety.
+
+A good SeaTunnel test is deterministic (the same result on every run),
leak-free (it releases every
+resource it opens), and cheap (it starts the fewest containers possible). The
guidelines below apply these
+principles to the two kinds of tests SeaTunnel contributors write.
+
+SeaTunnel distinguishes two test layers:
+
+- **Unit tests** (`*Test`, run by the Surefire plugin) validate connector or
core logic in isolation,
+ without external systems. See [Unit Test Guidelines](#unit-test-guidelines).
+- **End-to-End (E2E) tests** (`*IT`, run by the Failsafe plugin) validate
source, transform, and sink
+ behavior against real services in Testcontainers. See [E2E Test
Guidelines](#e2e-test-guidelines).
+
+When a pull request changes both logic and integration behavior, add tests in
both layers and run both
+before submitting.
+
+## Unit Test Guidelines
+
+### 1. Test behavior and contract, not implementation details
+
+A unit test should verify input-output behavior and option contracts. Avoid
testing private
+method internals or temporary implementation structure.
+
+Real example from `JdbcSourceFactoryTest` (module: `connector-jdbc`):
+
+```java
+private Map<String, Object> baseConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:mysql://localhost:3306/test");
+ cfg.put("driver", "com.mysql.cj.jdbc.Driver");
+ return cfg;
+}
+
+@Test
+void testValidConfigWithTablePath() {
+ Map<String, Object> cfg = baseConfig();
+ cfg.put("table_path", "test.users");
+ Assertions.assertDoesNotThrow(() -> validate(cfg));
+}
+```
+
+This test validates the external configuration contract and remains stable
even if internal factory
+implementation changes.
+
+### 2. Keep unit tests deterministic and local
+
+Unit tests should run fast and deterministically:
+
+- no `Thread.sleep`
+- no random assertions without fixed seed
+- no dependency on wall-clock timing
+
+If a case introduces cross-process or external dependencies, clearly classify
it in the proper test
+layer and keep assertions deterministic and diagnosable.
+
+### 3. Isolate dependencies with mock, stub, or fake
+
+A unit test should isolate the target behavior from external IO. Use in-memory
or lightweight
+test doubles for collaborators when possible.
+
+Keep one assertion scope per test method, so failures point to a single
behavior.
+
+Real engine-side mock example from `JobInfoServiceNullSafetyTest` (module:
`seatunnel-engine-server`):
+
+```java
+@BeforeEach
+void setUp() {
+ nodeEngine = mock(NodeEngineImpl.class);
+ hazelcastInstance = mock(HazelcastInstance.class);
+ runningJobInfoMap = mock(IMap.class);
+ finishedJobStateMap = mock(IMap.class);
+ finishedJobMetricsMap = mock(IMap.class);
+ finishedJobVertexInfoMap = mock(IMap.class);
+
+ when(nodeEngine.getHazelcastInstance()).thenReturn(hazelcastInstance);
+
when(hazelcastInstance.getMap(Constant.IMAP_RUNNING_JOB_INFO)).thenReturn(runningJobInfoMap);
+
when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_STATE)).thenReturn(finishedJobStateMap);
+ when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_METRICS))
+ .thenReturn(finishedJobMetricsMap);
+ when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO))
+ .thenReturn(finishedJobVertexInfoMap);
+
+ jobInfoService = new JobInfoService(nodeEngine);
+}
+
+@Test
+void shouldReturnJobIdOnlyWhenFinishedMetricsIsMissing() {
+ when(runningJobInfoMap.get(jobId)).thenReturn(null);
+ when(finishedJobStateMap.get(jobId)).thenReturn(jobState);
+ when(finishedJobMetricsMap.get(jobId)).thenReturn(null);
+
+ JsonObject result = jobInfoService.getJobInfoJson(jobId);
+ Assertions.assertEquals(jobId.toString(),
result.getString(RestConstant.JOB_ID, null));
+}
+```
+
+This example illustrates three practices:
+
+- It mocks every external map dependency at the boundary instead of starting
the engine services.
+- It uses `when(...).thenReturn(...)` to shape a single business branch per
test.
+- Its assertions target the observable output (the result JSON contract), not
internal calls.
+
+### 4. Verify exception type and message for invalid input paths
+
+For negative paths, assert both exception class and core message, so the test
protects user-facing
+error quality.
+
+Real example from `MongodbIncrementalSourceFactoryTest` (module:
`connector-cdc-mongodb`):
+
+```java
+Assertions.assertThrows(
+ MongodbConnectorException.class,
+ () ->
+ MongodbSourceConfigProvider.newBuilder()
+ .startupOptions(
+ new StartupConfig(StartupMode.EARLIEST, null,
null, null)));
+```
+
+### 5. Use clear naming and Arrange-Act-Assert structure
+
+- Name classes as `*Test` and methods as explicit behavior statements.
+- Keep each test in Arrange-Act-Assert order.
+- Avoid large helper logic inside test methods; extract reusable setup into
focused helpers.
+
+## Unit Test Checklist
+
+- [ ] Class name follows `*Test`; method names describe behavior or error case
+- [ ] Unit tests keep deterministic execution and avoid non-essential runtime
dependencies
+- [ ] Assertions target behavior and contract, not internal implementation
details
+- [ ] Negative tests assert both exception type and key error message
+- [ ] Test data setup is minimal and reusable
+
+## E2E Test Guidelines
+
+An E2E test extends `TestSuiteBase`, runs the connector inside a
Testcontainers-managed engine container, and
+verifies the result of a real job. The six rules below keep such tests
deterministic, leak-free, and cheap.
+
+### 1. Use Dynamic Ports, Never Hardcode
+
+Testcontainers assigns a random host port at startup. Hardcoding a port causes
conflicts in CI, where the
+port may already be taken or multiple suites run in parallel.
+
+Resolve the host port from the container at runtime:
+
+```java
+// Bad — collides in CI
+String brokerUrl = "tcp://127.0.0.1:61616";
+
+// Good — let Testcontainers assign the host port
+String brokerUrl = "tcp://" + container.getHost() + ":" +
container.getMappedPort(61616);
+config.setPort(container.getFirstMappedPort()); // when only one port is
exposed
+```
+
+Apply the same rule to all external services:
+
+```java
+// Database
+String jdbcUrl = String.format(
+ "jdbc:mysql://%s:%d/%s",
+ mysqlContainer.getHost(), mysqlContainer.getMappedPort(3306),
DATABASE);
+
+// HTTP service
+String endpoint = String.format(
+ "http://%s:%d";, serviceContainer.getHost(),
serviceContainer.getMappedPort(8080));
+```
+
+:::caution Job config files are the exception
+
+The SeaTunnel job runs inside the Docker network, so its config (`.conf`) must
reference the container's
+network alias (`withNetworkAliases("activemq-host")`) and the internal port
(e.g. `61616`), not the
+mapped host port.
+
+:::
+
+### 2. Wait on Conditions, Never `Thread.sleep`
+
+`Thread.sleep` is non-deterministic: too short and the test becomes flaky, too
long and CI is slowed down. It
+also produces no useful error when the expected state never arrives. Use
+[Awaitility](https://github.com/awaitility/awaitility) to poll for the actual
condition instead.
+
+```java
+// Bad — flaky, wasteful, and silent on failure
+container.executeJob("/job.conf");
+Thread.sleep(30000);
+Assertions.assertIterableEquals(expected, query());
+
+// Good — returns as soon as the condition holds, fails with the last
assertion error
+Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .pollDelay(Duration.ZERO)
+ .untilAsserted(() -> Assertions.assertIterableEquals(expected,
query()));
+```
+
+Choose the timeout from the scenario rather than picking a round number:
+
+| Scenario | `atMost` | `pollInterval` | Why
|
+|-----------------------------------|------------|----------------|--------------------------------------|
+| Container / client readiness | 2 min | 1 s | Image pull
+ service init is slow |
+| Job reaches `RUNNING` state | 1 min | 2 s | Scheduling
overhead |
+| Batch job result verified | 60 s | 2 s | Small
dataset completion |
+| Kafka / MQ message consumption | 30–60 s | 1 s | Consumer
group rebalance |
+| CDC / schema-change propagation | 60–120 s | 2–5 s | Binlog
latency + snapshot |
+
+When the client throws until the service is ready, add `.ignoreExceptions()`:
+
+```java
+Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(this::initProducer);
+```
+
+A reusable helper for the common "wait for job RUNNING" step:
+
+```java
+private void awaitJobRunning(TestContainer container, String jobId) {
+ Awaitility.await()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(1, TimeUnit.MINUTES)
+ .untilAsserted(
+ () -> Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId)));
+}
+```
+
+### 3. Release Resources Promptly
+
+A leaked connection exhausts container resources and available host ports, and
can cause CI to hang. Every
+resource you open must be closed. Most E2E tests implement the `TestResource`
interface and override its
+`tearDown()` method, which runs once after all tests in the class. Close
resources there in reverse order of
+creation, with a null check on each (the test may have failed before the
resource was created), and stop any
+manually-started container explicitly.
+
+```java
+@AfterAll
+@Override
+public void tearDown() throws Exception {
+ if (producer != null) {
+ producer.close();
+ }
+ if (session != null) {
+ session.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ if (container != null) {
+ container.stop();
+ }
+}
+```
+
+For resources scoped to a single method, prefer try-with-resources over manual
close:
+
+```java
+private void executeDml(String sql) {
+ try (Connection conn = getJdbcConnection();
+ Statement stmt = conn.createStatement()) {
+ stmt.execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException("Execute DML failed: " + sql, e);
+ }
+}
+```
+
+Cleanup checklist:
+
+- [ ] JDBC connections / statements closed
+- [ ] Message broker connections, sessions, producers, consumers closed
+- [ ] Custom clients (HTTP, gRPC) shut down
+- [ ] `ExecutorService` shut down with a timeout
+- [ ] Async `CompletableFuture` jobs cancelled
+- [ ] Temporary files / directories deleted
+- [ ] Manually-created Docker networks removed
+
+### 4. Submit Long-Running Jobs Asynchronously
+
+A streaming or CDC job runs until it is cancelled, so calling `executeJob`
inline blocks the test thread
+indefinitely and leaves no opportunity to inject data or assert intermediate
state. Submit such jobs with
+`CompletableFuture.supplyAsync`, wait until the job reaches `RUNNING`, run the
test steps, verify the result,
+then cancel the job.
+
+```java
+String jobId = "streaming-cdc-job";
+
+CompletableFuture<Void> job = CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.executeJob("/streaming_job.conf", jobId);
+ } catch (Exception e) {
+ log.error("Job execution failed", e);
+ throw new CompletionException(e); // propagate, never swallow
+ }
+ return null;
+ });
+
+awaitJobRunning(container, jobId); // wait for RUNNING before proceeding
+
+insertCdcData(); // run test steps while the job is running
+
+Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> verifySinkResults());
+
+container.cancelJob(jobId); // cancel the job
+```
+
+Use this table to decide:
+
+| Job type | Pattern
|
+|--------------------------------------------|----------------------------------|
+| Streaming job with data injection | `CompletableFuture.supplyAsync`
|
+| Streaming CDC job | `CompletableFuture.supplyAsync`
|
+| Dynamic partition discovery | `CompletableFuture.supplyAsync`
|
+| Batch job (source → sink that completes) | Inline `executeJob`
|
+
+For a batch job, assert the exit code and verify the result:
+
+```java
+Container.ExecResult result = container.executeJob("/batch_job.conf");
+Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() ->
verifyResults());
+```
+
+### 5. Keep Tests Deterministic and Cheap
+
+- Extend `TestSuiteBase` and implement `TestResource` to gain the `startUp()`
/ `tearDown()` lifecycle hooks.
+- Use `@TestTemplate` with a `TestContainer` parameter so the test runs across
every configured engine.
+- Start the fewest containers possible. Put the source and sink cases in the
same class so the Docker
+ image initializes once (see Coding Guide item 11).
+- Cover the full data type set, not just `String` and `int`.
+- Order dependent steps with
`@TestMethodOrder(MethodOrderer.OrderAnnotation.class)` and `@Order(n)`.
+- Add `@Slf4j` and log the key phases (data insertion, verification) to make
CI failures easier to diagnose.
+
+### 6. Whitelist Unrecyclable Client Threads (Zeta Engine)
+
+After the last job in a test finishes, the Zeta `SeaTunnelContainer` snapshots
the server JVM's live threads
+and fails the test if any non-system thread is still running after a
120-second grace window. This guards
+against connectors that leak threads. Most client threads die once you close
the client in `tearDown()`
+(Rule 3) — always try that first.
+
+However, some third-party client libraries (JDBC drivers, HTTP clients,
connection pools) start daemon
+threads that are never recycled and cannot be closed from the test. These
would fail the leak check through
+no fault of the test. For those — and only those — whitelist the thread's name
prefix by extending
+`isIssueWeAlreadyKnow(String)` in `SeaTunnelContainer`, where the project
keeps these exceptions centrally:
+
+```java
+//
seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/
+// container/seatunnel/SeaTunnelContainer.java
+
+/** The thread should be recycled but not, we should fix it in the future. */
+protected boolean isIssueWeAlreadyKnow(String threadName) {
+ return threadName.startsWith("ClickHouseClientWorker") // ClickHouse client
+ || threadName.startsWith("Okio Watchdog") // InfluxDB / OkHttp
+ || threadName.startsWith("iceberg-worker-pool") // Iceberg worker
pool
+ // Add the new connector's thread prefix here, with a comment
naming the library.
+ || threadName.startsWith("<your-connector-thread-prefix>");
+}
+```
+
+When adding an entry:
+
+- Match a specific prefix, never a broad substring — a loose match can hide a
real leak from another test.
+- Add a comment naming the connector and the fully-qualified class that spawns
the thread, following the
+ existing style.
+- Only whitelist threads that genuinely cannot be closed. A thread you *can*
close belongs in `tearDown()`,
+ not the whitelist.
+
+:::note
+
+Engine and JVM system threads (`hz.main`, `seatunnel-coordinator-service`,
`pool-N-thread-N`, GC threads, and
+so on) are already excluded by `isSystemThread(...)`. Do not duplicate them in
`isIssueWeAlreadyKnow(...)`.
+
+:::
+
+### Putting It Together
+
+The example below applies the rules together: a single shared container on a
pinned image, a dynamic host
+port, condition-based waiting during start-up, asynchronous job handling, and
resource cleanup in
+`tearDown()`.
+
+```java
+@Slf4j
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class ExampleConnectorIT extends TestSuiteBase implements TestResource {
+
+ private static final String CONTAINER_IMAGE =
"example/example-server:1.2.3";
+ private static final String CONTAINER_HOST = "example-host";
+ private static final int CONTAINER_PORT = 8080;
+
+ private GenericContainer<?> container;
+ private Connection connection;
+
+ @BeforeAll
+ @Override
+ public void startUp() throws Exception {
+ container =
+ new GenericContainer<>(DockerImageName.parse(CONTAINER_IMAGE))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(CONTAINER_HOST)
+ .withExposedPorts(CONTAINER_PORT)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(CONTAINER_IMAGE)));
+ Startables.deepStart(Stream.of(container)).join();
+ log.info("Example container started");
+
+ // Wait until the client can connect, rather than sleeping for a fixed
time.
+ Awaitility.given()
+ .ignoreExceptions()
+ .await()
+ .atMost(2, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(this::initConnection);
+ initializeTestData();
+ }
+
+ @AfterAll
+ @Override
+ public void tearDown() throws Exception {
+ if (connection != null) {
+ connection.close();
+ }
+ if (container != null) {
+ container.stop();
+ }
+ }
+
+ @TestTemplate
+ @Order(1)
+ public void testBatchSourceToSink(TestContainer container) throws
Exception {
+ Container.ExecResult result =
container.executeJob("/example_batch.conf");
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(this::verifyResults);
+ }
+
+ private void initConnection() {
+ String endpoint =
+ String.format(
+ "http://%s:%d";, container.getHost(),
container.getMappedPort(CONTAINER_PORT));
+ connection = createConnection(endpoint);
+ }
+}
+```
+
+## Module and Resource Layout
+
+A connector E2E test is its own Maven module under
`seatunnel-e2e/seatunnel-connector-v2-e2e/`. The engine
+loads `.conf` and other resource files from the test classpath root
(`src/test/resources/`), so these
+paths are conventions to follow, not arbitrary choices.
+
+```
+seatunnel-e2e/seatunnel-connector-v2-e2e/
+└── connector-<name>-e2e/ # one module per
connector
+ ├── pom.xml # artifactId:
connector-<name>-e2e
+ └── src/test/
+ ├── java/org/apache/seatunnel/e2e/connector/<name>/
+ │ └── <Name>IT.java # the IT class
+ └── resources/
+ ├── <name>_source_to_sink.conf # job configs
(classpath root)
+ ├── ddl/<table>.sql # schema / seed
DDL (optional)
+ ├── docker/<server>.cnf # container config
to mount (optional)
+ └── log4j2-test.properties # test logging
(optional)
+```
+
+### Test class
+
+- Location: `src/test/java/org/apache/seatunnel/e2e/connector/<name>/`. CDC
connectors instead use the
+ `org.apache.seatunnel.connectors.seatunnel.cdc.<name>` package — follow the
sibling CDC modules.
+- Naming: name the class `<Name>IT`. The Failsafe plugin runs `*IT` classes as
integration tests, while
+ Surefire runs `*Test` classes as unit tests. Use one class per behavior,
e.g. `<Name>WithSchemaChangeIT`.
+
+### Config (`.conf`) files
+
+- Place job configs directly under `src/test/resources/`, and reference them
from code with a leading
+ slash: `container.executeJob("/<name>_source_to_sink.conf")`. The leading
slash means classpath-absolute,
+ not a filesystem path.
+- Name files by intent and data flow, for example `fake_to_<sink>.conf`,
`<name>_multitable.conf`, or
+ `<name>cdc_to_<name>_with_schema_change.conf`. Avoid generic names such as
`test.conf`.
+- Inside the `.conf`, services must use the container's network alias and
internal port
+ (e.g. `host = "rabbitmq-e2e"`, `port = 5672`), never the mapped host port
(see Rule 1).
+- A `.conf` file is a source file and must carry the Apache License header as
`#` comments.
+
+### DDL / SQL and mounted resources
+
+- Put schema and seed scripts under `src/test/resources/ddl/`, named after the
database or table they build
+ (e.g. `ddl/inventory.sql`). CDC helpers such as `UniqueDatabase` resolve
them by the classpath name
+ `ddl/<template>.sql`, so keep that exact path.
+- Files mounted into a container (e.g. `my.cnf`, `setup.sql`) go under
`src/test/resources/docker/` and are
+ referenced through the container API (e.g.
`.withSetupSQL("docker/setup.sql")`).
+- Every new `.sql` file must also carry the Apache License header.
+
+### Registering the module
+
+A new module is not built until it is wired in:
+
+1. Add `<module>connector-<name>-e2e</module>` to
`seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml`.
+2. In the module `pom.xml`, set the `artifactId` to `connector-<name>-e2e`,
give it a `<name>` such as
+ `SeaTunnel : E2E : Connector V2 : <Name>`, and declare test dependencies on
the connector under test plus
+ `connector-console`, `connector-assert`, or `connector-fake` as required by
the job configs.
+3. For a brand-new connector, also update the CI label and
`plugin-mapping.properties` (see the
+ contribution checklist).
+
+## Verify Before Submitting
+
+```bash
+# Format
+./mvnw spotless:apply
+
+# Run a specific E2E test class (run more than once to confirm stability)
+./mvnw -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-<name>-e2e \
+ -DskipUT -DskipIT=false -Dit.test=<TestClassName> verify
+```
+
+## Checklist
+
+- [ ] UT follows `*Test` naming, deterministic execution, and dependency
isolation
+- [ ] UT negative paths assert exception type and key error message
+- [ ] No hardcoded host ports — all use `getMappedPort()` /
`getFirstMappedPort()`
+- [ ] No `Thread.sleep()` — all waits use Awaitility with scenario-based
timeouts
+- [ ] Every opened resource is closed in `@AfterAll` / `@AfterEach` or
try-with-resources
+- [ ] Streaming / CDC jobs submitted via `CompletableFuture.supplyAsync` and
gated on `RUNNING`
+- [ ] Source and sink share one container; full data types covered
+- [ ] Resources follow the layout: `.conf` at the classpath root, DDL under
`ddl/`, mounts under `docker/`
+- [ ] New module registered in the parent `pom.xml`
+- [ ] No Zeta thread-leak failure; any unrecyclable client thread whitelisted
in `isIssueWeAlreadyKnow`
+- [ ] Apache License header present on all new files, including `.conf` and
`.sql`
diff --git a/docs/sidebars.js b/docs/sidebars.js
index eda13f8e29..1e70330204 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -338,6 +338,7 @@ const sidebars = {
"developer/setup",
"developer/contribution-path",
"developer/coding-guide",
+ "developer/test-coding-guide",
"developer/how-to-create-your-connector",
"developer/source-connector-development",
"developer/sink-connector-development",
diff --git a/docs/zh/developer/test-coding-guide.md
b/docs/zh/developer/test-coding-guide.md
new file mode 100644
index 0000000000..8aaff9d7d8
--- /dev/null
+++ b/docs/zh/developer/test-coding-guide.md
@@ -0,0 +1,518 @@
+---
+title: 测试编码指南
+---
+
+# 测试编码指南
+
+本指南介绍如何为 Apache SeaTunnel 编写一个高质量、稳定的端到端（E2E）测试或单元测试。
+它是对[编码指南](coding-guide.md)的补充：编码指南覆盖通用的 PR 质量要求，而本指南专注于测试的稳定性与资源安全。
+
+一个优秀的 SeaTunnel 测试应当是确定性的（每次运行结果一致）、无泄漏的（释放所有打开的资源）、
+低成本的（启动尽可能少的容器）。下面的规范将这些原则落实到 SeaTunnel 贡献者编写的两类测试上。
+
+SeaTunnel 将测试分为两层：
+
+- **单元测试**（`*Test`，由 Surefire 插件运行）在隔离环境中校验 connector 或 core 逻辑，不依赖外部系统。
+ 参见[单元测试规范](#单元测试规范)。
+- **端到端（E2E）测试**（`*IT`，由 Failsafe 插件运行）借助 Testcontainers，针对真实服务校验 source、
+ transform、sink 的行为。参见 [E2E 测试规范](#e2e-测试规范)。
+
+当一个 Pull Request 同时改动了逻辑与集成行为时，请在两层都补充测试，并在提交前分别运行。
+
+## 单元测试规范
+
+### 1. 测行为和契约，不测实现细节
+
+单元测试应验证输入输出行为和配置契约，不要绑定私有方法内部实现或临时代码结构。
+
+真实案例：`JdbcSourceFactoryTest`（模块：`connector-jdbc`）
+
+```java
+private Map<String, Object> baseConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:mysql://localhost:3306/test");
+ cfg.put("driver", "com.mysql.cj.jdbc.Driver");
+ return cfg;
+}
+
+@Test
+void testValidConfigWithTablePath() {
+ Map<String, Object> cfg = baseConfig();
+ cfg.put("table_path", "test.users");
+ Assertions.assertDoesNotThrow(() -> validate(cfg));
+}
+```
+
+这个测试保护的是对外配置契约，即使工厂内部实现重构，测试仍然稳定。
+
+### 2. 单元测试必须确定且本地可跑
+
+单元测试应快速且确定：
+
+- 不使用 `Thread.sleep`
+- 不使用无固定种子的随机断言
+- 不依赖墙钟时间
+
+如果用例引入跨进程或外部依赖，请明确归类到对应测试层级，并保证断言可重复、可诊断。
+
+### 3. 用 mock、stub、fake 隔离依赖
+
+单元测试要把目标行为与外部 IO 隔离。对协作依赖优先使用内存实现或轻量测试替身。
+
+每个测试方法尽量只覆盖一个断言范围，失败时能直接定位单一行为。
+
+引擎侧真实 mock 案例：`JobInfoServiceNullSafetyTest`（模块：`seatunnel-engine-server`）
+
+```java
+@BeforeEach
+void setUp() {
+ nodeEngine = mock(NodeEngineImpl.class);
+ hazelcastInstance = mock(HazelcastInstance.class);
+ runningJobInfoMap = mock(IMap.class);
+ finishedJobStateMap = mock(IMap.class);
+ finishedJobMetricsMap = mock(IMap.class);
+ finishedJobVertexInfoMap = mock(IMap.class);
+
+ when(nodeEngine.getHazelcastInstance()).thenReturn(hazelcastInstance);
+
when(hazelcastInstance.getMap(Constant.IMAP_RUNNING_JOB_INFO)).thenReturn(runningJobInfoMap);
+
when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_STATE)).thenReturn(finishedJobStateMap);
+ when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_METRICS))
+ .thenReturn(finishedJobMetricsMap);
+ when(hazelcastInstance.getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO))
+ .thenReturn(finishedJobVertexInfoMap);
+
+ jobInfoService = new JobInfoService(nodeEngine);
+}
+
+@Test
+void shouldReturnJobIdOnlyWhenFinishedMetricsIsMissing() {
+ when(runningJobInfoMap.get(jobId)).thenReturn(null);
+ when(finishedJobStateMap.get(jobId)).thenReturn(jobState);
+ when(finishedJobMetricsMap.get(jobId)).thenReturn(null);
+
+ JsonObject result = jobInfoService.getJobInfoJson(jobId);
+ Assertions.assertEquals(jobId.toString(),
result.getString(RestConstant.JOB_ID, null));
+}
+```
+
+该示例体现了三项实践：
+
+- 在边界处 mock 所有外部 map 依赖，无需启动引擎服务；
+- 通过 `when(...).thenReturn(...)` 精准构造单一业务分支；
+- 断言只关注可观察输出（结果 JSON 契约），而非内部调用。
+
+### 4. 错误路径同时校验异常类型与关键报错信息
+
+负向用例不仅要断言异常类，还要断言关键报错内容，以保护用户可见错误质量。
+
+真实案例：`MongodbIncrementalSourceFactoryTest`（模块：`connector-cdc-mongodb`）
+
+```java
+Assertions.assertThrows(
+ MongodbConnectorException.class,
+ () ->
+ MongodbSourceConfigProvider.newBuilder()
+ .startupOptions(
+ new StartupConfig(StartupMode.EARLIEST, null,
null, null)));
+```
+
+### 5. 命名清晰，结构采用 Arrange-Act-Assert
+
+- 类名使用 `*Test`，方法名明确表达行为或错误场景。
+- 每个测试按 Arrange-Act-Assert 组织。
+- 避免在测试方法里写大量准备逻辑，可提取成小型复用 helper。
+
+## 单元测试检查清单
+
+- [ ] 类名符合 `*Test`，方法名可直接表达行为或错误场景
+- [ ] 单元测试保持确定性执行，避免不必要的运行时依赖
+- [ ] 断言目标是行为和契约，而非实现细节
+- [ ] 负向用例同时断言异常类型和关键报错信息
+- [ ] 测试数据准备最小且可复用
+
+## E2E 测试规范
+
+一个 E2E 测试继承 `TestSuiteBase`，在 Testcontainers 管理的引擎容器内运行 connector，并校验真实作业的结果。
+下面六条规则用于保证此类测试的确定性、无泄漏与低成本。
+
+### 1. 使用动态端口，禁止硬编码
+
+Testcontainers 在启动时会分配一个随机的宿主机端口。硬编码端口会在 CI 中引发冲突——端口可能已被占用，
+或者多个测试套件并行运行。
+
+应在运行时从容器解析宿主机端口：
+
+```java
+// Bad — collides in CI
+String brokerUrl = "tcp://127.0.0.1:61616";
+
+// Good — let Testcontainers assign the host port
+String brokerUrl = "tcp://" + container.getHost() + ":" +
container.getMappedPort(61616);
+config.setPort(container.getFirstMappedPort()); // when only one port is
exposed
+```
+
+对所有外部服务都适用此规则：
+
+```java
+// Database
+String jdbcUrl = String.format(
+ "jdbc:mysql://%s:%d/%s",
+ mysqlContainer.getHost(), mysqlContainer.getMappedPort(3306),
DATABASE);
+
+// HTTP service
+String endpoint = String.format(
+ "http://%s:%d";, serviceContainer.getHost(),
serviceContainer.getMappedPort(8080));
+```
+
+:::caution 作业配置文件是个例外
+
+SeaTunnel 作业运行在 Docker 网络内部，因此其配置（`.conf`）必须引用容器的网络别名
+（`withNetworkAliases("activemq-host")`）和内部端口（如 `61616`），而不是映射后的宿主机端口。
+
+:::
+
+### 2. 基于条件等待，禁止 `Thread.sleep`
+
+`Thread.sleep` 是非确定性的：时间太短测试会不稳定，太长则拖慢 CI。当期望状态始终未到达时，它也无法给出有用的
+错误信息。应使用 [Awaitility](https://github.com/awaitility/awaitility) 轮询真实条件。
+
+```java
+// Bad — flaky, wasteful, and silent on failure
+container.executeJob("/job.conf");
+Thread.sleep(30000);
+Assertions.assertIterableEquals(expected, query());
+
+// Good — returns as soon as the condition holds, fails with the last
assertion error
+Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .pollDelay(Duration.ZERO)
+ .untilAsserted(() -> Assertions.assertIterableEquals(expected,
query()));
+```
+
+应根据场景选择超时时间，而非随手填一个整数：
+
+| 场景 | `atMost` | `pollInterval` | 原因
|
+|---------------------------|------------|----------------|--------------------------|
+| 容器 / 客户端就绪 | 2 分钟 | 1 秒 | 镜像拉取 + 服务初始化较慢 |
+| 作业进入 `RUNNING` 状态 | 1 分钟 | 2 秒 | 调度开销
|
+| 批作业结果校验 | 60 秒 | 2 秒 | 小数据集完成 |
+| Kafka / MQ 消息消费 | 30–60 秒 | 1 秒 | 消费组再平衡
|
+| CDC / Schema 变更传播 | 60–120 秒 | 2–5 秒 | binlog 延迟 + 快照
|
+
+当客户端在服务就绪前会抛异常时，加上 `.ignoreExceptions()`：
+
+```java
+Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(this::initProducer);
+```
+
+一个可复用的「等待作业 RUNNING」辅助方法：
+
+```java
+private void awaitJobRunning(TestContainer container, String jobId) {
+ Awaitility.await()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(1, TimeUnit.MINUTES)
+ .untilAsserted(
+ () -> Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId)));
+}
+```
+
+### 3. 及时释放资源
+
+泄漏的连接会耗尽容器资源和宿主机可用端口，并可能导致 CI 挂起。你打开的每一个资源都必须关闭。大多数 E2E
+测试会实现 `TestResource` 接口并重写其 `tearDown()` 方法，该方法在类中所有测试结束后执行一次。应在其中按
+创建的逆序关闭资源，并对每个资源做 null 检查（测试可能在资源创建前就已失败），同时显式停止手动启动的容器。
+
+```java
+@AfterAll
+@Override
+public void tearDown() throws Exception {
+ if (producer != null) {
+ producer.close();
+ }
+ if (session != null) {
+ session.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ if (container != null) {
+ container.stop();
+ }
+}
+```
+
+对于仅在单个方法内使用的资源，优先使用 try-with-resources 而非手动 close：
+
+```java
+private void executeDml(String sql) {
+ try (Connection conn = getJdbcConnection();
+ Statement stmt = conn.createStatement()) {
+ stmt.execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException("Execute DML failed: " + sql, e);
+ }
+}
+```
+
+资源清理检查清单：
+
+- [ ] JDBC 连接 / Statement 已关闭
+- [ ] 消息中间件的 connection、session、producer、consumer 已关闭
+- [ ] 自定义客户端（HTTP、gRPC）已关闭
+- [ ] `ExecutorService` 带超时关闭
+- [ ] 异步的 `CompletableFuture` 作业已取消
+- [ ] 临时文件 / 目录已删除
+- [ ] 手动创建的 Docker 网络已移除
+
+### 4. 异步提交长时间运行的作业
+
+流作业或 CDC 作业会一直运行直到被取消，因此内联调用 `executeJob` 会使测试线程无限期阻塞，也就没有机会注入
+数据或断言中间状态。这类作业应使用 `CompletableFuture.supplyAsync` 提交，等待作业进入 `RUNNING` 状态，执行
+测试步骤，校验结果，最后取消作业。
+
+```java
+String jobId = "streaming-cdc-job";
+
+CompletableFuture<Void> job = CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.executeJob("/streaming_job.conf", jobId);
+ } catch (Exception e) {
+ log.error("Job execution failed", e);
+ throw new CompletionException(e); // propagate, never swallow
+ }
+ return null;
+ });
+
+awaitJobRunning(container, jobId); // wait for RUNNING before proceeding
+
+insertCdcData(); // run test steps while the job is running
+
+Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> verifySinkResults());
+
+container.cancelJob(jobId); // cancel the job
+```
+
+用下表判断：
+
+| 作业类型 | 模式 |
+|---------------------------------|----------------------------------|
+| 带数据注入的流作业 | `CompletableFuture.supplyAsync` |
+| 流式 CDC 作业 | `CompletableFuture.supplyAsync` |
+| 动态分区发现 | `CompletableFuture.supplyAsync` |
+| 批作业（source → sink 会自然结束） | 内联 `executeJob` |
+
+对于批作业，断言退出码并校验结果：
+
+```java
+Container.ExecResult result = container.executeJob("/batch_job.conf");
+Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() ->
verifyResults());
+```
+
+### 5. 保持测试确定且低成本
+
+- 继承 `TestSuiteBase` 并实现 `TestResource` 接口，以获得 `startUp()` / `tearDown()`
生命周期钩子。
+- 使用带 `TestContainer` 参数的 `@TestTemplate`，使测试在每个已配置的引擎上运行。
+- 启动尽可能少的容器。把 source 和 sink 的用例放在同一个类中，让 Docker 镜像只初始化一次
+ （参见编码指南第 11 条）。
+- 覆盖完整的数据类型集合，不要只测 `String` 和 `int`。
+- 用 `@TestMethodOrder(MethodOrderer.OrderAnnotation.class)` 和 `@Order(n)`
对有依赖的步骤排序。
+- 加上 `@Slf4j` 并记录关键阶段（数据插入、校验）的日志，便于排查 CI 失败。
+
+### 6. 将无法回收的客户端线程加入白名单（Zeta 引擎）
+
+测试中最后一个作业结束后，Zeta 的 `SeaTunnelContainer` 会对服务端 JVM 的存活线程做一次快照，如果在 120 秒
+宽限期之后仍有非系统线程在运行，就会让测试失败。这是为了防止 connector 泄漏线程。大多数客户端线程在你于
+`tearDown()` 中关闭客户端后（规则 3）就会消亡——请务必先尝试这种方式。
+
+但有些第三方客户端库（JDBC 驱动、HTTP 客户端、连接池）会启动永远不会被回收、且测试无法关闭的守护线程。这类
+线程会让泄漏检查失败，而这并非测试本身的问题。仅针对这种情况，可以通过扩展 `SeaTunnelContainer` 中的
+`isIssueWeAlreadyKnow(String)` 方法，把该线程的名称前缀加入白名单——项目将这些例外集中维护于此：
+
+```java
+//
seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/
+// container/seatunnel/SeaTunnelContainer.java
+
+/** The thread should be recycled but not, we should fix it in the future. */
+protected boolean isIssueWeAlreadyKnow(String threadName) {
+ return threadName.startsWith("ClickHouseClientWorker") // ClickHouse client
+ || threadName.startsWith("Okio Watchdog") // InfluxDB / OkHttp
+ || threadName.startsWith("iceberg-worker-pool") // Iceberg worker
pool
+ // Add the new connector's thread prefix here, with a comment
naming the library.
+ || threadName.startsWith("<your-connector-thread-prefix>");
+}
+```
+
+添加白名单条目时：
+
+- 匹配具体的前缀，绝不要用宽泛的子串——过松的匹配会掩盖其他测试的真实泄漏。
+- 按现有风格加上注释，说明所属 connector 以及启动该线程的全限定类名。
+- 只对确实无法关闭的线程加白名单。能够关闭的线程应放在 `tearDown()` 中处理，而不是白名单。
+
+:::note
+
+引擎与 JVM 系统线程（`hz.main`、`seatunnel-coordinator-service`、`pool-N-thread-N`、GC
线程等）已由
+`isSystemThread(...)` 排除，请勿在 `isIssueWeAlreadyKnow(...)` 中重复添加。
+
+:::
+
+### 整体示例
+
+下面的示例综合应用了上述规则：测试方法共用一个固定版本镜像的容器、使用动态宿主机端口、在启动阶段基于
+条件等待、异步处理作业，并在 `tearDown()` 中清理资源。
+
+```java
+@Slf4j
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class ExampleConnectorIT extends TestSuiteBase implements TestResource {
+
+ private static final String CONTAINER_IMAGE =
"example/example-server:1.2.3";
+ private static final String CONTAINER_HOST = "example-host";
+ private static final int CONTAINER_PORT = 8080;
+
+ private GenericContainer<?> container;
+ private Connection connection;
+
+ @BeforeAll
+ @Override
+ public void startUp() throws Exception {
+ container =
+ new GenericContainer<>(DockerImageName.parse(CONTAINER_IMAGE))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(CONTAINER_HOST)
+ .withExposedPorts(CONTAINER_PORT)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(CONTAINER_IMAGE)));
+ Startables.deepStart(Stream.of(container)).join();
+ log.info("Example container started");
+
+ // Wait until the client can connect, rather than sleeping for a fixed
time.
+ Awaitility.given()
+ .ignoreExceptions()
+ .await()
+ .atMost(2, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(this::initConnection);
+ initializeTestData();
+ }
+
+ @AfterAll
+ @Override
+ public void tearDown() throws Exception {
+ if (connection != null) {
+ connection.close();
+ }
+ if (container != null) {
+ container.stop();
+ }
+ }
+
+ @TestTemplate
+ @Order(1)
+ public void testBatchSourceToSink(TestContainer container) throws
Exception {
+ Container.ExecResult result =
container.executeJob("/example_batch.conf");
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(this::verifyResults);
+ }
+
+ private void initConnection() {
+ String endpoint =
+ String.format(
+ "http://%s:%d";, container.getHost(),
container.getMappedPort(CONTAINER_PORT));
+ connection = createConnection(endpoint);
+ }
+}
+```
+
+## 模块与资源结构
+
+一个 connector 的 E2E 测试是位于 `seatunnel-e2e/seatunnel-connector-v2-e2e/` 下的独立
Maven 模块。引擎从测试的
+classpath 根目录（`src/test/resources/`）加载 `.conf` 及其他资源文件，因此这些路径是必须遵循的约定，而非随意
+摆放。
+
+```
+seatunnel-e2e/seatunnel-connector-v2-e2e/
+└── connector-<name>-e2e/ # 每个 connector 一个模块
+ ├── pom.xml # artifactId:
connector-<name>-e2e
+ └── src/test/
+ ├── java/org/apache/seatunnel/e2e/connector/<name>/
+ │ └── <Name>IT.java # IT 测试类
+ └── resources/
+ ├── <name>_source_to_sink.conf # 作业配置（classpath
根目录）
+ ├── ddl/<table>.sql # 建表 / 初始化 DDL（可选）
+ ├── docker/<server>.cnf # 挂载到容器的配置（可选）
+ └── log4j2-test.properties # 测试日志（可选）
+```
+
+### 测试类
+
+- 位置： `src/test/java/org/apache/seatunnel/e2e/connector/<name>/`。CDC connector
改用
+ `org.apache.seatunnel.connectors.seatunnel.cdc.<name>` 包——请与同级的 CDC 模块保持一致。
+- 命名： 测试类命名为 `<Name>IT`。Failsafe 插件以集成测试方式运行 `*IT` 类，而 Surefire 以单元测试方式运行
+ `*Test` 类。每个行为一个类，例如 `<Name>WithSchemaChangeIT`。
+
+### 配置（`.conf`）文件
+
+- 作业配置直接放在 `src/test/resources/` 下，并在代码中用前导斜线引用：
+ `container.executeJob("/<name>_source_to_sink.conf")`。前导斜线表示 classpath
绝对路径，而非文件系统路径。
+- 按用途与数据流向命名，例如 `fake_to_<sink>.conf`、`<name>_multitable.conf` 或
+ `<name>cdc_to_<name>_with_schema_change.conf`。避免使用 `test.conf` 这类泛化名称。
+- 在 `.conf` 内部，服务必须使用容器的网络别名和内部端口（如 `host = "rabbitmq-e2e"`、`port = 5672`），
+ 而不是映射后的宿主机端口（参见规则 1）。
+- `.conf` 文件属于源文件，必须以 `#` 注释形式带上 Apache License 头。
+
+### DDL / SQL 及挂载资源
+
+- 建表与初始化脚本放在 `src/test/resources/ddl/` 下，按其所建的数据库或表命名（如 `ddl/inventory.sql`）。
+ `UniqueDatabase` 等 CDC 辅助类按 classpath 名称 `ddl/<template>.sql` 解析，请保持该路径不变。
+- 挂载到容器的文件（如 `my.cnf`、`setup.sql`）放在 `src/test/resources/docker/` 下，通过容器 API 引用
+ （如 `.withSetupSQL("docker/setup.sql")`）。
+- 每个新增的 `.sql` 文件同样必须带上 Apache License 头。
+
+### 注册模块
+
+新模块在被纳入构建前不会被编译：
+
+1. 在 `seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml` 中加入
`<module>connector-<name>-e2e</module>`。
+2. 在模块的 `pom.xml` 中将 `artifactId` 设为 `connector-<name>-e2e`，给定形如
+ `SeaTunnel : E2E : Connector V2 : <Name>` 的 `<name>`，并按作业配置所需声明对被测
connector 以及
+ `connector-console`、`connector-assert`、`connector-fake` 的测试依赖。
+3. 对于全新的 connector，还需更新 CI label 和 `plugin-mapping.properties`（参见贡献检查清单）。
+
+## 提交前验证
+
+```bash
+# Format
+./mvnw spotless:apply
+
+# Run a specific E2E test class (run more than once to confirm stability)
+./mvnw -pl seatunnel-e2e/seatunnel-connector-v2-e2e/connector-<name>-e2e \
+ -DskipUT -DskipIT=false -Dit.test=<TestClassName> verify
+```
+
+## 检查清单
+
+- [ ] UT 遵循 `*Test` 命名、确定性执行和依赖隔离
+- [ ] UT 负向用例同时断言异常类型和关键报错信息
+- [ ] 无硬编码宿主机端口——全部使用 `getMappedPort()` / `getFirstMappedPort()`
+- [ ] 无 `Thread.sleep()`——所有等待都使用 Awaitility，并按场景设定超时
+- [ ] 每个打开的资源都在 `@AfterAll` / `@AfterEach` 或 try-with-resources 中关闭
+- [ ] 流 / CDC 作业通过 `CompletableFuture.supplyAsync` 提交，并以 `RUNNING` 为前置条件
+- [ ] source 与 sink 共用一个容器；覆盖完整数据类型
+- [ ] 资源遵循约定结构：`.conf` 位于 classpath 根目录，DDL 在 `ddl/` 下，挂载文件在 `docker/` 下
+- [ ] 新模块已在父 `pom.xml` 中注册
+- [ ] 无 Zeta 线程泄漏失败；无法回收的客户端线程已在 `isIssueWeAlreadyKnow` 中加入白名单
+- [ ] 所有新文件（含 `.conf` 和 `.sql`）都带有 Apache License 头
