This is an automated email from the ASF dual-hosted git repository.
zhangshenghang 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 847b51e390 [Fix][E2E] Split unrelated connector CI fixes from DB2 CDC
PR (#10925)
847b51e390 is described below
commit 847b51e39092c377572d01fbb8d931a3d75334bc
Author: David Zollo <[email protected]>
AuthorDate: Fri Aug 21 20:28:55 2026 +0800
[Fix][E2E] Split unrelated connector CI fixes from DB2 CDC PR (#10925)
Co-authored-by: DanielLeens <[email protected]>
Co-authored-by: Daniel <[email protected]>
Co-authored-by: DanielLeens <[email protected]>
---
.../cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java | 29 ++-
.../e2e/connector/file/ftp/FtpFileIT.java | 281 ++++++++++++++++-----
...ftp_file_json_to_assert_with_multipletable.conf | 6 +-
.../e2e/connector/file/fstp/SftpFileIT.java | 173 ++++++++-----
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 58 ++++-
.../seatunnel/e2e/connector/kafka/KafkaITTest.java | 38 +++
.../e2e/connector/kafka/KafkaKerberosIT.java | 24 ++
.../kafka_to_kafka_exactly_once_streaming.conf | 13 +-
.../e2e/connector/rocketmq/RocketMqIT.java | 64 ++++-
.../rocketmq_source_group_offset_to_console.conf | 5 +-
.../rocketmq/rocketmq_source_restore.conf | 6 +-
.../e2e/connector/tdengine/TDengineIT.java | 80 +++---
12 files changed, 575 insertions(+), 202 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
index 025dcc36d1..344fd79135 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
@@ -67,12 +67,17 @@ import static org.awaitility.Awaitility.await;
public class MysqlCDCWithFlinkSchemaChangeIT extends TestSuiteBase implements
TestResource {
/**
* Flink schema evolution can restart after XA recover/rollback on loaded
CI runners, so these
- * assertions need enough time for the job to recover and replay the
schema-change event.
+ * assertions need enough time for the job to recover and replay the
schema-change event. The
+ * long bound is a failure deadline, not the expected recovery duration.
*/
- private static final long SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS =
300_000L;
+ private static final long SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS =
600_000L;
private static final long STRUCTURE_AND_DATA_ASSERT_TIMEOUT_MILLIS =
300_000L;
- private static final int MAX_TIMESTAMP_DRIFT_SECONDS = 60;
+ /**
+ * The timestamp default is evaluated by different MySQL statements during
CDC replay, so this
+ * tolerance allows loaded CI scheduling jitter without hiding
minute-level CDC lag.
+ */
+ private static final int MAX_TIMESTAMP_DRIFT_SECONDS = 45;
private static final String MYSQL_DATABASE = "shop";
private static final String SOURCE_TABLE = "products";
@@ -202,9 +207,9 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
query(String.format(PROJECTION_QUERY,
database, sourceTable)),
query(String.format(PROJECTION_QUERY,
database, sinkTable)));
- // The default value of add_column4 is
current_timestamp()，so the
- // history data of sink table with this column may
be different from the
- // source table because delay of apply schema
change.
+ // The default value of add_column4 is
current_timestamp(), so the
+ // history data of the sink table can lag briefly
behind the source
+ // table while the schema-change event is still
propagating.
String query =
String.format(
"SELECT t1.id AS table1_id,
t1.add_column4 AS table1_timestamp, "
@@ -272,12 +277,14 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
// case1 add columns with cdc data at same time
shopDatabase.setTemplateName("add_columns").createAndInitialize();
- await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
+ await().pollInterval(5, TimeUnit.SECONDS)
+ .atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertSchemaDescriptionEqualsIgnoringColumnOrder(
database, sourceTable, sinkTable));
- await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
+ await().pollInterval(5, TimeUnit.SECONDS)
+ .atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
assertTableDataEqualsBySourceColumnOrder(
@@ -287,9 +294,9 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
query(String.format(PROJECTION_QUERY,
database, sourceTable)),
query(String.format(PROJECTION_QUERY,
database, sinkTable)));
- // The default value of add_column4 is
current_timestamp()，so the
- // history data of sink table with this column may
be different from the
- // source table because delay of apply schema
change.
+ // The default value of add_column4 is
current_timestamp(), so the
+ // history data of the sink table can lag briefly
behind the source
+ // table while the schema-change event is still
propagating.
String query =
String.format(
"SELECT t1.id AS table1_id,
t1.add_column4 AS table1_timestamp, "
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java
index 7075de2c0a..d23fe15e50 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java
@@ -249,10 +249,11 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void testFtpBinaryUpdateModeContinuousDiscoveryDistcp(TestContainer
container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath(CONTINUOUS_DISTCP_PATH);
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putFtpFile(CONTINUOUS_DISTCP_PATH + "/src/test1.bin", "abc");
@@ -297,13 +298,12 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
"def",
readFtpFile(
CONTINUOUS_DISTCP_PATH +
"/dst/test2.bin")));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(ftpHomeDir + CONTINUOUS_DISTCP_PATH);
+ cleanupContinuousJob(
+ container, jobId, jobFuture, ftpHomeDir +
CONTINUOUS_DISTCP_PATH, testFailure);
}
}
@@ -313,10 +313,11 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testFtpBinaryUpdateModeContinuousDiscoveryPostSyncDelete(TestContainer
container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath(CONTINUOUS_DELETE_PATH);
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putFtpFile(CONTINUOUS_DELETE_PATH + "/src/delete-test.bin", "abc");
@@ -353,13 +354,12 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
CONTINUOUS_DELETE_PATH
+
"/src/delete-test.bin"),
"source file should be deleted
after checkpoint-gated post-sync commit"));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(ftpHomeDir + CONTINUOUS_DELETE_PATH);
+ cleanupContinuousJob(
+ container, jobId, jobFuture, ftpHomeDir +
CONTINUOUS_DELETE_PATH, testFailure);
}
}
@@ -369,10 +369,11 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testFtpBinaryUpdateModeContinuousDiscoveryPostSyncBackup(TestContainer
container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath(CONTINUOUS_BACKUP_PATH);
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putFtpFile(CONTINUOUS_BACKUP_PATH + "/src/backup-test.bin", "abc");
@@ -422,13 +423,12 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
"backup-test.bin.v*")
> 0,
"backup target should contain
version-suffixed file"));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(ftpHomeDir + CONTINUOUS_BACKUP_PATH);
+ cleanupContinuousJob(
+ container, jobId, jobFuture, ftpHomeDir +
CONTINUOUS_BACKUP_PATH, testFailure);
}
}
@@ -437,11 +437,11 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
value = {},
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
- public void testFtpContinuousBackupRetentionCleanup(TestContainer
container)
- throws IOException, InterruptedException {
+ public void testFtpContinuousBackupRetentionCleanup(TestContainer
container) throws Throwable {
resetContinuousTestPath(CONTINUOUS_RETENTION_PATH);
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putFtpFile(CONTINUOUS_RETENTION_PATH + "/src/retention-input.bin",
"input");
putFtpFile(CONTINUOUS_RETENTION_PATH +
"/backup/retention-old.bin.v3_123456", "abc");
@@ -472,27 +472,35 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
CONTINUOUS_RETENTION_PATH
+ "/backup",
"retention-old.bin.v*"),
"retention should remove expired
SeaTunnel backup files"));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(ftpHomeDir + CONTINUOUS_RETENTION_PATH);
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ ftpHomeDir + CONTINUOUS_RETENTION_PATH,
+ testFailure);
}
}
+ /**
+ * Verifies continuous non-recursive FTP sync copies root files while
ignoring nested files.
+ *
+ * <p>The nested-file assertion protects the connector's non-recursive
discovery contract.
+ */
@TestTemplate
@DisabledOnContainer(
value = {},
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void testFtpBinaryUpdateModeContinuousDiscoveryWithNonRecursiveScan(
- TestContainer container) throws IOException, InterruptedException {
+ TestContainer container) throws Throwable {
resetContinuousTestPath(CONTINUOUS_NON_RECURSIVE_PATH);
-
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putFtpFile(CONTINUOUS_NON_RECURSIVE_PATH + "/src/root.bin",
"root");
putFtpFile(CONTINUOUS_NON_RECURSIVE_PATH +
"/src/subdir/nested.bin", "nested");
@@ -523,53 +531,149 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
Thread.sleep(3000);
Assertions.assertFalse(
isFtpFileExists(CONTINUOUS_NON_RECURSIVE_PATH +
"/dst/subdir/nested.bin"));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(ftpHomeDir +
CONTINUOUS_NON_RECURSIVE_PATH);
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ ftpHomeDir + CONTINUOUS_NON_RECURSIVE_PATH,
+ testFailure);
}
}
+ /**
+ * Verifies non-recursive FTP distcp updates root files without
overwriting nested files.
+ *
+ * <p>The stale nested destination file must remain unchanged after the
job finishes.
+ */
@TestTemplate
public void
testFtpBinaryUpdateModeDistcpWithNonRecursiveScan(TestContainer container)
throws IOException, InterruptedException {
resetUpdateTestPath();
- putFtpFile("/tmp/seatunnel/update/src/root.bin", "root-updated-v2");
- putFtpFile("/tmp/seatunnel/update/src/subdir/nested.bin",
"nest-updated-v2");
- putFtpFile("/tmp/seatunnel/update/dst/root.bin", "root-stale-v1");
- putFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin",
"nest-stale-v1");
-
- Container.ExecResult execResult =
-
container.executeJob("/text/ftp_binary_update_non_recursive_distcp.conf");
- Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
- Assertions.assertEquals(
- "root-updated-v2",
readFtpFile("/tmp/seatunnel/update/dst/root.bin"));
- Assertions.assertEquals(
- "nest-stale-v1",
readFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin"));
+ try {
+ putFtpFile("/tmp/seatunnel/update/src/root.bin",
"root-updated-v2");
+ putFtpFile("/tmp/seatunnel/update/src/subdir/nested.bin",
"nest-updated-v2");
+ putFtpFile("/tmp/seatunnel/update/dst/root.bin", "root-stale-v1");
+ putFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin",
"nest-stale-v1");
- deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/update");
+ Container.ExecResult execResult =
+
container.executeJob("/text/ftp_binary_update_non_recursive_distcp.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ Assertions.assertEquals(
+ "root-updated-v2",
readFtpFile("/tmp/seatunnel/update/dst/root.bin"));
+ Assertions.assertEquals(
+ "nest-stale-v1",
readFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin"));
+ } finally {
+ deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/update");
+ }
}
+ /**
+ * Verifies strict checksum mode keeps nested FTP files untouched during
non-recursive scans.
+ */
@TestTemplate
public void
testFtpBinaryUpdateModeStrictChecksumSkipsNestedChangesWithNonRecursiveScan(
TestContainer container) throws IOException, InterruptedException {
resetUpdateTestPath();
- putFtpFile("/tmp/seatunnel/update/src/root.bin", "root-same-v1");
- putFtpFile("/tmp/seatunnel/update/src/subdir/nested.bin",
"nest-new-v1");
- putFtpFile("/tmp/seatunnel/update/dst/root.bin", "root-same-v1");
- putFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin",
"nest-old-v1");
+ try {
+ putFtpFile("/tmp/seatunnel/update/src/root.bin", "root-same-v1");
+ putFtpFile("/tmp/seatunnel/update/src/subdir/nested.bin",
"nest-new-v1");
+ putFtpFile("/tmp/seatunnel/update/dst/root.bin", "root-same-v1");
+ putFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin",
"nest-old-v1");
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/text/ftp_binary_update_non_recursive_strict_checksum.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ Assertions.assertEquals(
+ "root-same-v1",
readFtpFile("/tmp/seatunnel/update/dst/root.bin"));
+ Assertions.assertEquals(
+ "nest-old-v1",
readFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin"));
+ } finally {
+ deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/update");
+ }
+ }
- Container.ExecResult execResult =
-
container.executeJob("/text/ftp_binary_update_non_recursive_strict_checksum.conf");
- Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
- Assertions.assertEquals("root-same-v1",
readFtpFile("/tmp/seatunnel/update/dst/root.bin"));
- Assertions.assertEquals(
- "nest-old-v1",
readFtpFile("/tmp/seatunnel/update/dst/subdir/nested.bin"));
+ /**
+ * Continuous discovery is only considered stopped after the engine
reaches CANCELED and the
+ * submit command exits, so cleanup regressions still fail this E2E test.
+ */
+ private void assertContinuousJobStopsAfterCancel(
+ TestContainer container,
+ String jobId,
+ CompletableFuture<Container.ExecResult> jobFuture)
+ throws IOException, InterruptedException {
+ if (jobFuture == null) {
+ return;
+ }
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Container.ExecResult cancelResult =
container.cancelJob(jobId);
+ Assertions.assertEquals(
+ 0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ });
- deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/update");
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "CANCELED",
+ container.getJobStatus(jobId),
+ "Continuous job should be canceled
before the test exits."));
+
+ Awaitility.await()
+ .atMost(180, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .until(jobFuture::isDone);
+
+ try {
+ Container.ExecResult execResult = jobFuture.get(30,
TimeUnit.SECONDS);
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ } catch (Exception e) {
+ throw new RuntimeException("Wait continuous job exit failed.", e);
+ }
+ }
+
+ /**
+ * Stops a continuous job before deleting its shared test path, without
hiding the original test
+ * failure when cleanup also fails.
+ */
+ private void cleanupContinuousJob(
+ TestContainer container,
+ String jobId,
+ CompletableFuture<Container.ExecResult> jobFuture,
+ String cleanupPath,
+ Throwable testFailure)
+ throws Throwable {
+ Throwable cleanupFailure = null;
+ try {
+ assertContinuousJobStopsAfterCancel(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ cleanupFailure = failure;
+ }
+
+ if (cleanupFailure == null) {
+ try {
+ deleteFileFromContainer(cleanupPath);
+ } catch (Throwable failure) {
+ cleanupFailure = failure;
+ }
+ }
+
+ if (cleanupFailure == null) {
+ return;
+ }
+ if (testFailure != null) {
+ testFailure.addSuppressed(cleanupFailure);
+ } else {
+ throw cleanupFailure;
+ }
}
@TestTemplate
@@ -664,6 +768,19 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
// test read recursive file path
helper.execute("/text/ftp_file_text_recursive_to_assert.conf");
helper.execute("/text/ftp_file_text_non_recursive_to_assert.conf");
+
+ String homePath = ftpHomeDir;
+ String sink01 = "/tmp/seatunnel/json/sink/multiplesource/fake01";
+ String sink02 = "/tmp/seatunnel/json/sink/multiplesource/fake02";
+ deleteFileFromContainer(homePath + sink01);
+ deleteFileFromContainer(homePath + sink02);
+ // Keep a dedicated source file for each logical table. Sharing one
FTP path between the
+ // two multiple-table entries can leave the second reader with no
physical file to open on
+ // slower CI runs.
+ ensureMultipleTableJsonInputFiles();
+
helper.execute("/json/ftp_file_json_to_assert_with_multipletable.conf");
+ Assertions.assertEquals(getFileListFromContainer(homePath +
sink01).size(), 1);
+ Assertions.assertEquals(getFileListFromContainer(homePath +
sink02).size(), 1);
}
@TestTemplate
@@ -819,6 +936,11 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
return catResult.getStdout() == null ? "" :
catResult.getStdout().trim();
}
+ /**
+ * Checks whether a file exists in the FTP container without creating
parent directories.
+ *
+ * <p>This helper is used by negative assertions where creating the path
would hide regressions.
+ */
private boolean isFtpFileExists(String ftpPath) throws IOException,
InterruptedException {
String containerPath = ftpHomeDir + ftpPath;
Container.ExecResult result =
@@ -984,6 +1106,10 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
}
private void ensureReadJsonInputFile() throws IOException,
InterruptedException {
+ // Reset the shared JSON source directory for each template run. The
FTP container is
+ // reused across engine variants, so stale fake01/fake02 inputs from
the multiple-table
+ // case can otherwise leak into the single-table assert and inflate
the row count.
+ deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/read/json");
Container.ExecResult mkdirResult =
ftpContainer.execInContainer(
"sh",
@@ -1002,6 +1128,39 @@ public class FtpFileIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(0, chmodResult.getExitCode(),
chmodResult.getStderr());
}
+ /**
+ * The multiple-table FTP JSON test reads two logical tables from two
table configs. Give each
+ * table its own physical source file so one reader never depends on a
file already consumed by
+ * the other table path. This helper owns {@code tmp/seatunnel/read/json}
for the suite and
+ * rebuilds it before the multiple-table run.
+ */
+ private void ensureMultipleTableJsonInputFiles() throws IOException,
InterruptedException {
+ // Rebuild the multiple-table source tree from scratch so previous
template runs never
+ // leave extra JSON files behind for the next engine invocation.
+ deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/read/json");
+ copyJsonInputFileTo(
+ ftpHomeDir +
"/tmp/seatunnel/read/json/fake01/name=tyrantlucifer/hobby=coding");
+ copyJsonInputFileTo(
+ ftpHomeDir +
"/tmp/seatunnel/read/json/fake02/name=tyrantlucifer/hobby=coding");
+ Container.ExecResult chmodResult =
+ ftpContainer.execInContainer(
+ "sh", "-c", "chmod -R 777 " + ftpHomeDir +
"/tmp/seatunnel/read/json");
+ Assertions.assertEquals(0, chmodResult.getExitCode(),
chmodResult.getStderr());
+ }
+
+ /**
+ * Copies the canonical JSON fixture into an isolated multiple-table input
directory.
+ *
+ * @param directory destination directory inside the FTP container
+ */
+ private void copyJsonInputFileTo(String directory) throws IOException,
InterruptedException {
+ Container.ExecResult mkdirResult =
+ ftpContainer.execInContainer("sh", "-c", "mkdir -p " +
directory);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+ ContainerUtil.copyFileIntoContainers(
+ "/json/e2e.json", directory + "/e2e.json", ftpContainer);
+ }
+
@SneakyThrows
private List<String> getFileListFromContainer(String path) {
String command = "ls -1 " + path;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/resources/json/ftp_file_json_to_assert_with_multipletable.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/resources/json/ftp_file_json_to_assert_with_multipletable.conf
index eb5cf0b7e6..9bbcf39a66 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/resources/json/ftp_file_json_to_assert_with_multipletable.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/resources/json/ftp_file_json_to_assert_with_multipletable.conf
@@ -35,7 +35,7 @@ source {
port = 21
user = seatunnel
password = pass
- path = "/tmp/seatunnel/read/json"
+ path = "/tmp/seatunnel/read/json/fake01"
file_format_type = "json"
schema = {
table = "fake01"
@@ -78,7 +78,7 @@ source {
port = 21
user = seatunnel
password = pass
- path = "/tmp/seatunnel/read/json"
+ path = "/tmp/seatunnel/read/json/fake02"
file_format_type = "json"
schema = {
table = "fake02"
@@ -137,4 +137,4 @@ sink {
filename_time_format = "yyyy.MM.dd"
is_enable_transaction = true
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
index f3a76c3dc4..8c3a49e560 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
@@ -285,10 +285,11 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testSftpBinaryUpdateModeContinuousDiscoveryDistcp(TestContainer container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath();
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putSftpFile(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous/src/test1.bin", "abc");
@@ -342,12 +343,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
SFTP_CONTAINER_HOME
+
"/tmp/seatunnel/continuous/dst/test2.bin")));
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous");
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ SFTP_CONTAINER_HOME + "/tmp/seatunnel/continuous",
+ testFailure);
}
}
@@ -357,10 +362,11 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testSftpBinaryUpdateModeContinuousDiscoveryWithNonRecursiveScan(
- TestContainer container) throws IOException, InterruptedException {
+ TestContainer container) throws Throwable {
resetContinuousTestPath();
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
jobFuture =
CompletableFuture.supplyAsync(
@@ -401,12 +407,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
SFTP_CONTAINER_HOME
+
"/tmp/seatunnel/continuous/dst/subdir/nested.bin")));
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous");
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ SFTP_CONTAINER_HOME + "/tmp/seatunnel/continuous",
+ testFailure);
}
}
@@ -416,10 +426,11 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testSftpBinaryUpdateModeContinuousDiscoveryPostSyncDelete(TestContainer
container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath();
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putSftpFile(
SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous/src/delete-test.bin", "abc");
@@ -458,12 +469,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
+
"/tmp/seatunnel/continuous/src/delete-test.bin"),
"source file should be deleted
after checkpoint-gated post-sync commit"));
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous");
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ SFTP_CONTAINER_HOME + "/tmp/seatunnel/continuous",
+ testFailure);
}
}
@@ -473,10 +488,11 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
public void
testSftpBinaryUpdateModeContinuousDiscoveryPostSyncBackup(TestContainer
container)
- throws IOException, InterruptedException {
+ throws Throwable {
resetContinuousTestPath();
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
putSftpFile(
SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous/src/backup-test.bin", "abc");
@@ -528,12 +544,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
> 0,
"backup target should contain
version-suffixed file"));
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous");
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ SFTP_CONTAINER_HOME + "/tmp/seatunnel/continuous",
+ testFailure);
}
}
@@ -542,11 +562,11 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
value = {},
type = {EngineType.FLINK, EngineType.SPARK},
disabledReason = "Continuous discovery is a long-running job; only
run in zeta engine.")
- public void testSftpContinuousBackupRetentionCleanup(TestContainer
container)
- throws IOException, InterruptedException {
+ public void testSftpContinuousBackupRetentionCleanup(TestContainer
container) throws Throwable {
resetContinuousTestPath();
String jobId = String.valueOf(JobIdGenerator.newJobId());
CompletableFuture<Container.ExecResult> jobFuture = null;
+ Throwable testFailure = null;
try {
String retentionFile =
SFTP_CONTAINER_HOME
@@ -579,12 +599,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
"retention-old.bin.v*"),
"retention should remove expired
SeaTunnel backup files"));
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
- waitContinuousJobExit(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
} finally {
- cancelContinuousJobQuietly(container, jobId, jobFuture);
- deleteFileFromContainer(SFTP_CONTAINER_HOME +
"/tmp/seatunnel/continuous");
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ SFTP_CONTAINER_HOME + "/tmp/seatunnel/continuous",
+ testFailure);
}
}
@@ -765,45 +789,34 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
}
/**
- * Best-effort cleanup for assertion failures so one continuous test
cannot leak into the next
- * test's shared SFTP directories.
+ * Continuous discovery is only considered stopped after the engine
reaches CANCELED and the
+ * submit command exits, so cleanup regressions still fail this E2E test.
*/
- private void cancelContinuousJobQuietly(
+ private void assertContinuousJobStopsAfterCancel(
TestContainer container,
String jobId,
- CompletableFuture<Container.ExecResult> jobFuture) {
- if (jobFuture == null || jobFuture.isDone()) {
+ CompletableFuture<Container.ExecResult> jobFuture)
+ throws IOException, InterruptedException {
+ if (jobFuture == null) {
return;
}
- try {
- String status = container.getJobStatus(jobId);
- if (!"CANCELED".equals(status)
- && !"FINISHED".equals(status)
- && !"FAILED".equals(status)) {
- container.cancelJob(jobId);
- }
- Awaitility.await()
- .atMost(180, TimeUnit.SECONDS)
- .pollInterval(2, TimeUnit.SECONDS)
- .until(jobFuture::isDone);
- } catch (Exception e) {
- log.warn("Failed to clean up continuous SFTP job {}.", jobId, e);
- }
- }
-
- /**
- * Wait for the continuous discovery job to enter the terminal canceled
state and for the
- * asynchronous execute call to finish its post-job thread cleanup.
- */
- private void waitContinuousJobExit(
- TestContainer container,
- String jobId,
- CompletableFuture<Container.ExecResult> jobFuture) {
Awaitility.await()
- .atMost(60, TimeUnit.SECONDS)
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Container.ExecResult cancelResult =
container.cancelJob(jobId);
+ Assertions.assertEquals(
+ 0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ });
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
.pollInterval(2, TimeUnit.SECONDS)
.untilAsserted(
- () -> Assertions.assertEquals("CANCELED",
container.getJobStatus(jobId)));
+ () ->
+ Assertions.assertEquals(
+ "CANCELED",
+ container.getJobStatus(jobId),
+ "Continuous job should be canceled
before the test exits."));
Awaitility.await()
.atMost(180, TimeUnit.SECONDS)
.pollInterval(2, TimeUnit.SECONDS)
@@ -816,6 +829,42 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
}
}
+ /**
+ * Stops a continuous job before deleting its shared test path, without
hiding the original test
+ * failure when cleanup also fails.
+ */
+ private void cleanupContinuousJob(
+ TestContainer container,
+ String jobId,
+ CompletableFuture<Container.ExecResult> jobFuture,
+ String cleanupPath,
+ Throwable testFailure)
+ throws Throwable {
+ Throwable cleanupFailure = null;
+ try {
+ assertContinuousJobStopsAfterCancel(container, jobId, jobFuture);
+ } catch (Throwable failure) {
+ cleanupFailure = failure;
+ }
+
+ if (cleanupFailure == null) {
+ try {
+ deleteFileFromContainer(cleanupPath);
+ } catch (Throwable failure) {
+ cleanupFailure = failure;
+ }
+ }
+
+ if (cleanupFailure == null) {
+ return;
+ }
+ if (testFailure != null) {
+ testFailure.addSuppressed(cleanupFailure);
+ } else {
+ throw cleanupFailure;
+ }
+ }
+
private void waitUntilContainerTimeAfter(long epochSeconds) {
Awaitility.await()
.atMost(10, TimeUnit.SECONDS)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index 626330bc8c..6c6730e0ef 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -145,7 +145,11 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
private List<ConsumerRecord<String, String>> nativeData;
- /** Topics created dynamically during tests; cleaned up in {@link
#tearDown()}. */
+ /**
+ * Topics created dynamically during tests.
+ *
+ * <p>They are cleaned up in {@link #tearDown()} to keep later Kafka E2E
cases isolated.
+ */
private final List<String> dynamicTopics = new CopyOnWriteArrayList<>();
private final AtomicInteger startModeTestSequence = new AtomicInteger();
@@ -1363,7 +1367,11 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.count();
}
- /** Get the current end offset (LEO) on partition-0. */
+ /**
+ * Gets the current end offset on partition-0.
+ *
+ * <p>The exactly-once assertions use this as the visible sink baseline.
+ */
private long endOffsetOnP0(String topic) {
try (KafkaConsumer<String, String> c = new
KafkaConsumer<>(kafkaConsumerConfig())) {
TopicPartition tp0 = new TopicPartition(topic, 0);
@@ -2517,7 +2525,11 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
createKafkaTopic(topicName);
}
- /** Build the dynamic `-i key=value` variables for the exactly-once
streaming template. */
+ /**
+ * Builds the dynamic {@code -i key=value} variables for the exactly-once
streaming template.
+ *
+ * <p>Each test run uses isolated topics and a dedicated consumer group.
+ */
private List<String> buildExactlyOnceStreamingVariables(
String sourceTopic, String sinkTopic, String consumerGroup) {
return Arrays.asList(
@@ -2785,29 +2797,49 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long visibleEndOffsetExclusive =
consumer.endOffsets(Collections.singletonList(topicPartition))
.get(topicPartition);
- long lastObservedOffset = startOffset - 1;
+ long nextOffset = startOffset;
int consecutiveEmptyPolls = 0;
do {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
if (records.isEmpty()) {
consecutiveEmptyPolls++;
- continue;
- }
- consecutiveEmptyPolls = 0;
- for (ConsumerRecord<String, String> record :
records.records(topicPartition)) {
- if (record.offset() >= startOffset &&
seenOffsets.add(record.offset())) {
- data.add(record.value());
+ } else {
+ consecutiveEmptyPolls = 0;
+ for (ConsumerRecord<String, String> record :
records.records(topicPartition)) {
+ if (record.offset() >= startOffset &&
seenOffsets.add(record.offset())) {
+ data.add(record.value());
+ }
}
- lastObservedOffset = Math.max(lastObservedOffset,
record.offset());
}
- } while (lastObservedOffset < visibleEndOffsetExclusive - 1
- && consecutiveEmptyPolls < 20);
+ long currentPosition = consumer.position(topicPartition);
+ // Exactly-once topics can contain transaction control records
or aborted records
+ // that advance offsets without surfacing visible
READ_COMMITTED records. Track the
+ // consumer position so we do not stop early while the broker
is still advancing the
+ // transactional scan range.
+ if (records.isEmpty()
+ && shouldStopAfterEmptyReadCommittedPoll(
+ currentPosition, nextOffset,
consecutiveEmptyPolls)) {
+ break;
+ }
+ nextOffset = currentPosition;
+ } while (nextOffset < visibleEndOffsetExclusive &&
consecutiveEmptyPolls < 20);
return data;
} finally {
closeKafkaConsumer(consumer);
}
}
+ /**
+ * Decides whether repeated empty READ_COMMITTED polls are stable enough
to stop scanning.
+ *
+ * <p>If the broker advances the consumer position across aborted or
control records, the scan
+ * must continue even when no visible records are returned.
+ */
+ static boolean shouldStopAfterEmptyReadCommittedPoll(
+ long currentPosition, long previousPosition, int
consecutiveEmptyPolls) {
+ return currentPosition == previousPosition && consecutiveEmptyPolls >=
20;
+ }
+
private Properties kafkaManualConsumerConfig() {
Properties props = kafkaConsumerConfig();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaITTest.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaITTest.java
new file mode 100644
index 0000000000..c30d0ec533
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaITTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.e2e.connector.kafka;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Verifies Kafka E2E helper behavior that is independent from the
Testcontainers runtime. */
+public class KafkaITTest {
+
+ /**
+ * READ_COMMITTED scans must continue across aborted or control records
that advance position.
+ */
+ @Test
+ public void shouldContinueEmptyReadCommittedPollWhenPositionAdvances() {
+ // Aborted transactions and control records can move the position
without returning visible
+ // READ_COMMITTED records, so the helper must continue scanning that
offset range.
+
Assertions.assertFalse(KafkaIT.shouldStopAfterEmptyReadCommittedPoll(11L, 10L,
20));
+
Assertions.assertFalse(KafkaIT.shouldStopAfterEmptyReadCommittedPoll(15L, 10L,
30));
+
Assertions.assertFalse(KafkaIT.shouldStopAfterEmptyReadCommittedPoll(10L, 10L,
19));
+
Assertions.assertTrue(KafkaIT.shouldStopAfterEmptyReadCommittedPoll(10L, 10L,
20));
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
index 9d24670b78..2c2f5c4609 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
@@ -103,6 +103,12 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
private GenericContainer<?> kafkaContainer;
private GenericContainer<?> kerberosContainer;
+ /** Original JVM Krb5 config property that must be restored after this
suite. */
+ private String previousKrb5Conf;
+
+ /** Original JVM JAAS config property that must be restored after this
suite. */
+ private String previousJaasConfig;
+
private final SeaTunnelRowType SEATUNNEL_ROW_TYPE =
new SeaTunnelRowType(
new String[] {
@@ -143,6 +149,8 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
@BeforeAll
@Override
public void startUp() throws Exception {
+ previousKrb5Conf = System.getProperty("java.security.krb5.conf");
+ previousJaasConfig =
System.getProperty("java.security.auth.login.config");
System.setProperty(
"java.security.krb5.conf",
ContainerUtil.getResourcesFile("/kerberos/krb5_local.conf").getPath());
@@ -479,5 +487,21 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
if (kerberosContainer != null) {
kerberosContainer.close();
}
+ // KafkaKerberosIT sets global JVM JAAS/Krb5 properties during
startup. Clear them after
+ // the suite so the plain Kafka E2E tests in the same Maven JVM do not
inherit Kerberos
+ // authentication settings by mistake.
+ restoreSystemProperty("java.security.krb5.conf", previousKrb5Conf);
+ restoreSystemProperty("java.security.auth.login.config",
previousJaasConfig);
+ }
+
+ /**
+ * Restores a JVM security property to the value that existed before this
test suite changed it.
+ */
+ private void restoreSystemProperty(String key, String previousValue) {
+ if (previousValue == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, previousValue);
+ }
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf
index ac82598d49..244bf7dcc0 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf
@@ -24,11 +24,16 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "${sourceTopic}"
- consumer.group = "${consumerGroup}"
+ topic = "${sourceTopic:kafka_topic_exactly_once_1}"
+ consumer.group = "${consumerGroup:test_exactly_once}"
# The default format is json, which is optional
format = text
- start_mode = group_offsets
+ # The exactly-once test writes a readiness probe before the job starts,
while the
+ # consumer group is still new and has no stored offsets. Read from
earliest so
+ # KafkaSourceSplitEnumerator can assign the initial split on the first run.
+ # Restore tests resume from checkpointed splits, so start_mode is not used
on
+ # the restored path.
+ start_mode = earliest
kafka.config = {
auto.offset.reset = "earliest"
enable.auto.commit = "false"
@@ -42,7 +47,7 @@ transform {}
sink{
kafka {
format = text
- topic = "${sinkTopic}"
+ topic = "${sinkTopic:kafka_topic_exactly_once_2}"
bootstrap.servers = "kafkaCluster:9092"
semantics = EXACTLY_ONCE
kafka.config = {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
index 8cabce57d1..ef62a04d68 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
@@ -382,15 +382,18 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
@TestTemplate
public void testSourceRocketMqStartConfig(TestContainer container)
throws IOException, InterruptedException {
+ final String topicName = "test_topic_group_" + uniqueTestSuffix();
+ final String consumerGroup = "SeaTunnel-Consumer-Group-" +
uniqueTestSuffix();
+
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
- "test_topic_group",
+ topicName,
null,
SEATUNNEL_ROW_TYPE,
DEFAULT_FORMAT,
DEFAULT_FIELD_DELIMITER);
- generateTestData(row -> serializer.serializeRow(row),
"test_topic_group", 100, 150);
- testRocketMqGroupOffsetsToConsole(container);
+ generateTestData(row -> serializer.serializeRow(row), topicName, 100,
150);
+ executeRocketMqGroupOffsetsToConsole(container, topicName,
consumerGroup);
}
@TestTemplate
@@ -412,10 +415,17 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(tag, data.get(key).getTag());
}
- public void testRocketMqGroupOffsetsToConsole(TestContainer container)
+ /**
+ * Uses isolated topic and consumer-group names so template invocations
cannot share offsets.
+ */
+ private void executeRocketMqGroupOffsetsToConsole(
+ TestContainer container, String topicName, String consumerGroup)
throws IOException, InterruptedException {
Container.ExecResult execResult =
-
container.executeJob("/rocketmq/rocketmq_source_group_offset_to_console.conf");
+ container.executeJob(
+
"/rocketmq/rocketmq_source_group_offset_to_console.conf",
+ Arrays.asList(
+ "sourceTopic=" + topicName, "consumerGroup=" +
consumerGroup));
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
}
@@ -606,10 +616,18 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
throws IOException, InterruptedException, MQBrokerException,
RemotingException,
MQClientException, ExecutionException {
- final String sourceTopic = "test_topic_restore";
- final String sinkTopic = "test_topic_restore_output";
+ final String uniqueSuffix = uniqueTestSuffix();
+ final String sourceTopic = "test_topic_restore_" + uniqueSuffix;
+ final String sinkTopic = "test_topic_restore_output_" + uniqueSuffix;
+ final String consumerGroup = "restore_consumer_group_" + uniqueSuffix;
final String payload = "Seatunnel RocketMQ Restore Test Data";
final String jobId = "20260416";
+ final String[] restoreVariables =
+ new String[] {
+ "sourceTopic=" + sourceTopic,
+ "sinkTopic=" + sinkTopic,
+ "consumerGroup=" + consumerGroup
+ };
for (int i = 0; i < 20; i++) {
Message msg = new Message(sourceTopic, (payload + "_initial_" +
i).getBytes());
@@ -624,7 +642,9 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
() -> {
try {
return container.executeJob(
-
"/rocketmq/rocketmq_source_restore.conf", jobId);
+
"/rocketmq/rocketmq_source_restore.conf",
+ jobId,
+ restoreVariables);
} catch (Exception e) {
log.error("First job execution exception", e);
throw new RuntimeException(e);
@@ -675,6 +695,10 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
producer.send(msg, new MessageQueue(sourceTopic,
RocketMqContainer.BROKER_NAME, 0));
}
+ Awaitility.await()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> getTopicMaxOffset(sourceTopic) >=
srcEndBeforeStart + 25);
long srcEndAfterAll = getTopicMaxOffset(sourceTopic);
Assertions.assertTrue(
srcEndAfterAll >= srcEndBeforeStart + 25,
@@ -684,7 +708,8 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
CompletableFuture.runAsync(
() -> {
try {
-
container.restoreJob("/rocketmq/rocketmq_source_restore.conf", jobId);
+ container.restoreJob(
+ "/rocketmq/rocketmq_source_restore.conf",
jobId, restoreVariables);
} catch (Exception e) {
log.error("Restore job execution exception", e);
throw new RuntimeException(e);
@@ -702,13 +727,19 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(
expectedTotal,
finalSinkOffset,
- "Sink offset mismatch after restore — possible duplicate
consumption. "
+ "Sink offset mismatch after restore - possible duplicate
consumption. "
+ "Expected: "
+ expectedTotal
+ ", actual: "
+ finalSinkOffset);
-
List<String> allSinkMessages = pollMessagesFromOffset(sinkTopic, 0);
+ Assertions.assertEquals(
+ expectedTotal,
+ allSinkMessages.size(),
+ "Unexpected sink message count after restore. Expected: "
+ + expectedTotal
+ + ", actual: "
+ + allSinkMessages.size());
long initialCount =
allSinkMessages.stream().filter(body ->
body.contains("_initial_")).count();
long additionalCount =
@@ -815,6 +846,17 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
}
}
+ /**
+ * Returns a collision-resistant suffix for broker resources shared by
template invocations.
+ * RocketMQ accepts independent topic creation requests, so no
synchronization is needed for
+ * UUID-backed resource names.
+ *
+ * @return UUID text without separators, suitable for topic and
consumer-group names
+ */
+ private String uniqueTestSuffix() {
+ return UUID.randomUUID().toString().replace("-", "");
+ }
+
private void deleteTopicIfExist(String topicName) {
DefaultMQAdminExt admin = new DefaultMQAdminExt();
admin.setInstanceName(UUID.randomUUID().toString());
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_group_offset_to_console.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_group_offset_to_console.conf
index f727cbd504..9cf1ec1422 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_group_offset_to_console.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_group_offset_to_console.conf
@@ -25,10 +25,11 @@ env {
source {
Rocketmq {
name.srv.addr = "rocketmq-e2e:9876"
- topics = "test_topic_group"
+ topics = "${sourceTopic:test_topic_group}"
plugin_output = "rocketmq_table"
format = json
start.mode = "CONSUME_FROM_GROUP_OFFSETS"
+ consumer.group = "${consumerGroup:SeaTunnel-Consumer-Group}"
schema = {
fields {
id = bigint
@@ -63,4 +64,4 @@ sink {
]
}
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_restore.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_restore.conf
index 26fd95ac4a..5bb4f05907 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_restore.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq/rocketmq_source_restore.conf
@@ -25,11 +25,11 @@ env {
source {
Rocketmq {
name.srv.addr = "rocketmq-e2e:9876"
- topics = "test_topic_restore"
+ topics = "${sourceTopic:test_topic_restore}"
plugin_output = "rocketmq_restore_table"
format = text
start.mode = "CONSUME_FROM_FIRST_OFFSET"
- consumer.group = "restore_consumer_group"
+ consumer.group = "${consumerGroup:restore_consumer_group}"
schema = {
fields {
content = string
@@ -44,7 +44,7 @@ transform {
sink {
Rocketmq {
name.srv.addr = "rocketmq-e2e:9876"
- topic = "test_topic_restore_output"
+ topic = "${sinkTopic:test_topic_restore_output}"
format = text
producer.send.sync = true
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
index 6c60ac6bf1..2b9590f7a5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
@@ -38,6 +38,7 @@ import lombok.extern.slf4j.Slf4j;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
import java.util.Arrays;
@@ -53,6 +54,14 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
private static final String NETWORK_ALIASES1 = "flink_e2e_tdengine_src";
private static final String NETWORK_ALIASES2 = "flink_e2e_tdengine_sink";
private static final int PORT = 6041;
+ /**
+ * Keeps the test database within the vnode budget that small CI machines
can provide.
+ *
+ * <p>This is purely a CI resource workaround. The STABLE-backed source
and sink queries in this
+ * suite do not assert per-vgroup parallelism, so a single vgroup is
expected to behave the same
+ * as the default multi-vgroup layout for the exercised SQL paths.
+ */
+ private static final String SINGLE_VGROUP_DATABASE_OPTIONS = " KEEP 3650
VGROUPS 1";
private GenericContainer<?> tdengineServer1;
private GenericContainer<?> tdengineServer2;
@@ -90,17 +99,7 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
log.info("TDengine container started");
connection1 = createConnect(tdengineServer1);
connection2 = createConnect(tdengineServer2);
- // wait for TDengine fully start
- given().ignoreExceptions()
- .await()
- .atLeast(100, TimeUnit.MILLISECONDS)
- .pollInterval(1, TimeUnit.SECONDS)
- .atMost(120, TimeUnit.SECONDS)
- .untilAsserted(
- () ->
- Assertions.assertEquals(
- Boolean.TRUE,
- connection1.isValid(100) &
connection2.isValid(100)));
+ waitForTDengineReady();
testDataCount = generateTestDataSet();
log.info("tdengine testDataCount=" + testDataCount); // rowCount=8
}
@@ -108,16 +107,16 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
@SneakyThrows
private int generateTestDataSet() {
int rowCount;
- waitForDatabaseReady(connection1, "CREATE DATABASE power KEEP 3650");
try (Statement stmt = connection1.createStatement()) {
+ stmt.execute("CREATE DATABASE power" +
SINGLE_VGROUP_DATABASE_OPTIONS);
stmt.execute(
"CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT,
voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
String sql = getSQL();
rowCount = stmt.executeUpdate(sql);
}
- waitForDatabaseReady(connection2, "CREATE DATABASE power2 KEEP 3650");
try (Statement stmt = connection2.createStatement()) {
+ stmt.execute("CREATE DATABASE power2" +
SINGLE_VGROUP_DATABASE_OPTIONS);
stmt.execute(
"CREATE STABLE power2.meters2 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
@@ -134,8 +133,8 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
"CREATE STABLE power2.meters4 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
}
- waitForDatabaseReady(connection2, "CREATE DATABASE power3 KEEP 3650");
try (Statement stmt = connection2.createStatement()) {
+ stmt.execute("CREATE DATABASE power3" +
SINGLE_VGROUP_DATABASE_OPTIONS);
stmt.execute(
"CREATE STABLE power3.meters5 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
@@ -143,24 +142,6 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
return rowCount;
}
- /**
- * Waits for TDengine dnode to be fully online before creating the
database. The REST adapter
- * (port 6041) may accept connections before the dnode registers with the
management node,
- * causing "Out of dnodes" errors on CREATE DATABASE.
- */
- private void waitForDatabaseReady(Connection connection, String
createDbSql) {
- given().ignoreExceptions()
- .await()
- .pollInterval(2, TimeUnit.SECONDS)
- .atMost(60, TimeUnit.SECONDS)
- .untilAsserted(
- () -> {
- try (Statement stmt =
connection.createStatement()) {
- stmt.execute(createDbSql);
- }
- });
- }
-
@TestTemplate
public void testTDengine(TestContainer container) throws Exception {
Container.ExecResult execResult =
@@ -224,6 +205,41 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
return conn;
}
+ /**
+ * Waits until TDengine accepts JDBC traffic and can allocate vgroups for
new databases.
+ *
+ * <p>Port readiness alone does not prove that TDengine can create the
test databases.
+ */
+ private void waitForTDengineReady() {
+ given().ignoreExceptions()
+ .await()
+ .pollInterval(1, TimeUnit.SECONDS)
+ .atMost(120, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(connection1.isValid(100));
+ Assertions.assertTrue(connection2.isValid(100));
+ verifyDatabaseCanBeCreated(connection1,
"seatunnel_ready_src");
+ verifyDatabaseCanBeCreated(connection2,
"seatunnel_ready_sink");
+ });
+ }
+
+ /**
+ * TDengine can report a healthy socket before the storage node is ready
to create databases.
+ *
+ * <p>The probe database is intentionally kept until the container stops.
Reusing the same probe
+ * avoids repeated create/drop churn while the service is still becoming
ready.
+ */
+ private void verifyDatabaseCanBeCreated(Connection connection, String
databaseName)
+ throws SQLException {
+ try (Statement statement = connection.createStatement()) {
+ statement.execute(
+ "CREATE DATABASE IF NOT EXISTS "
+ + databaseName
+ + SINGLE_VGROUP_DATABASE_OPTIONS);
+ }
+ }
+
@AfterAll
@Override
public void tearDown() throws Exception {
