davidzollo commented on code in PR #10925:
URL: https://github.com/apache/seatunnel/pull/10925#discussion_r3689159863
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java:
##########
@@ -2590,23 +2624,35 @@ private List<String> getKafkaConsumerListData(String
topicName, long startOffset
long visibleEndOffsetExclusive =
consumer.endOffsets(Collections.singletonList(topicPartition))
.get(topicPartition);
+ long nextOffset = startOffset;
Long lastProcessedOffset = startOffset - 1;
int consecutiveEmptyPolls = 0;
Review Comment:
Addressed in the current head. The polling decision is factored into
`KafkaIT.shouldStopAfterEmptyReadCommittedPoll`, and
`KafkaITTest.shouldContinueEmptyReadCommittedPollWhenPositionAdvances` covers
the aborted/control-record boundary: empty polls with an advancing partition
position continue scanning, while a stable position stops only after the
empty-poll threshold.
##########
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:
Addressed in the current head. The cleanup path now preserves both branches
explicitly: if the test body already failed, cleanup failure is added as
suppressed; if cleanup is the first failure, it is thrown. The helper comment
now documents that behavior. I kept the timeout budget unchanged to avoid
introducing a weaker cleanup gate in this stabilization PR; it can be tuned
separately if CI duration becomes the next bottleneck.
--
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]
