zhangshenghang commented on code in PR #10925:
URL: https://github.com/apache/seatunnel/pull/10925#discussion_r3675124598
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java:
##########
@@ -343,72 +350,162 @@ public void
testFtpBinaryUpdateModeContinuousDiscoveryWithNonRecursiveScan(
throw new RuntimeException(e);
}
});
-
- putFtpFile("/tmp/seatunnel/continuous/src/root.bin", "root");
- putFtpFile("/tmp/seatunnel/continuous/src/subdir/nested.bin",
"nested");
-
- Awaitility.await()
- .atMost(60, TimeUnit.SECONDS)
- .untilAsserted(
- () ->
- Assertions.assertEquals(
- "root",
-
readFtpFile("/tmp/seatunnel/continuous/dst/root.bin")));
-
- Thread.sleep(3000);
-
Assertions.assertFalse(isFtpFileExists("/tmp/seatunnel/continuous/dst/subdir/nested.bin"));
-
- Container.ExecResult cancelResult = container.cancelJob(jobId);
- Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
-
- Container.ExecResult execResult;
+ Throwable testFailure = null;
try {
- execResult = jobFuture.get(120, TimeUnit.SECONDS);
- } catch (Exception e) {
- throw new RuntimeException("Wait continuous job exit failed.", e);
+ putFtpFile("/tmp/seatunnel/continuous/src/root.bin", "root");
+ putFtpFile("/tmp/seatunnel/continuous/src/subdir/nested.bin",
"nested");
+
+ Awaitility.await()
+ .atMost(120, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "root",
+
readFtpFile("/tmp/seatunnel/continuous/dst/root.bin")));
+
+ Thread.sleep(3000);
+ Assertions.assertFalse(
+
isFtpFileExists("/tmp/seatunnel/continuous/dst/subdir/nested.bin"));
+ } catch (Throwable failure) {
+ testFailure = failure;
+ throw failure;
+ } finally {
+ cleanupContinuousJob(
+ container,
+ jobId,
+ jobFuture,
+ ftpHomeDir + "/tmp/seatunnel/continuous",
+ testFailure);
}
- Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
-
- deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/continuous");
}
+ /**
+ * 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"));
-
- deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/update");
+ 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");
+
+ 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 {
+ 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(
Review Comment:
**P2 — `cleanupContinuousJob` exception handling**
```java
if (testFailure != null) {
testFailure.addSuppressed(cleanupFailure);
} else {
throw cleanupFailure;
}
```
If `cleanupContinuousJob` itself is the first failure (i.e. the test body
succeeded but cleanup threw), we throw `cleanupFailure` and lose the original
`testFailure`. Here `testFailure` is null in that case so it's actually fine —
but please consider using `assertThrows`-style logic and double-check that
`Throwable#addSuppressed` is reachable when the cleanup is called from
`finally` (it is, but only when `testFailure` is non-null, i.e. the test body
threw first). That matches the comment ("without hiding the original test
failure"), but worth a unit test that exercises both branches.
Also, `assertContinuousJobStopsAfterCancel` waits up to 30s + 30s + 180s = 4
minutes per continuous discovery test. With 4 such tests per suite on FTP and 4
on SFTP, cleanup alone can add ~32 minutes in the worst case. If that becomes a
problem in CI, consider lowering the 180s timeout.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java:
##########
@@ -765,6 +884,38 @@ private void ensureReadJsonInputFile() throws IOException,
InterruptedException
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.
+ */
+ private void ensureMultipleTableJsonInputFiles() throws IOException,
InterruptedException {
Review Comment:
**P3 — `ensureMultipleTableJsonInputFiles` deletes shared dir, may affect
later runs**
```java
deleteFileFromContainer(ftpHomeDir + "/tmp/seatunnel/read/json");
```
This is fine because the FTP container is per-suite (`@BeforeAll` starts
one), but if a future test adds a sibling read under
`tmp/seatunnel/read/json/...` between the multiple-table run and `tearDown`,
the data will be wiped. Worth a Javadoc note that the helper owns
`tmp/seatunnel/read/json` for the lifetime of the suite.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java:
##########
@@ -416,10 +419,17 @@ public void testSinkRocketMqMessageTag(TestContainer
container)
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.
+ */
+ public void testRocketMqGroupOffsetsToConsole(
Review Comment:
**P3 — `testRocketMqGroupOffsetsToConsole` is `public` and `test`-prefixed
but not a `@TestTemplate`**
```java
public void testRocketMqGroupOffsetsToConsole(TestContainer container,
String topicName, String consumerGroup)
```
This is a helper invoked only by `testSourceRocketMqStartConfig` (line 400).
The `test` prefix and `public` visibility are misleading. Recommend either:
1. Rename to `runRocketMqGroupOffsetsToConsole(...)` (or
`executeRocketMqGroupOffsetsToConsole`), or
2. Make it `private` (preferred, since there's only one caller).
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java:
##########
@@ -806,6 +837,15 @@ private long getTopicMaxOffset(String topicName) {
}
}
+ /**
+ * Returns a collision-resistant suffix for broker resources shared by
template invocations.
+ *
+ * @return UUID text without separators, suitable for topic and
consumer-group names
+ */
+ private String uniqueTestSuffix() {
Review Comment:
**P3 — `uniqueTestSuffix()` collision risk is very low but worth noting**
UUID without hyphens gives 32 hex chars (~128 bits). With one call per test
invocation the chance of collision is negligible, but two callers in the same
test (`testSourceRocketMqStartConfig` and `testSourceRocketMqRestore`) each
call it once, so a single suite has at most 2 topics × 2 groups = 4 resources.
Fine.
Just confirm that `createTopic` (or RocketMQ auto-create) tolerates
concurrent creation — if not, consider synchronizing the UUID generation.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java:
##########
@@ -68,7 +68,7 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
* 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.
*/
- 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;
Review Comment:
**P3 — `MAX_TIMESTAMP_DRIFT_SECONDS = 60` may be too lenient**
```java
private static final int MAX_TIMESTAMP_DRIFT_SECONDS = 60;
```
`add_column4` defaults to `current_timestamp()`. A 60-second drift between
source and sink is large enough to hide real CDC lag issues. Suggest 10-20
seconds once the CI is stable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
