davidzollo commented on code in PR #10925:
URL: https://github.com/apache/seatunnel/pull/10925#discussion_r3683943170
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf:
##########
@@ -24,11 +24,13 @@ 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 sends a readiness probe while the job starts.
+ # Read from earliest so a probe sent before split assignment is not
skipped.
+ start_mode = earliest
Review Comment:
Updated the PR description to document why this config uses `start_mode =
earliest`: the exactly-once test starts with a brand-new consumer group, so
`group_offsets` has no stored offsets and the source would not assign the
initial split. The config already notes that restore resumes from checkpointed
splits, so `start_mode` is not used on the restored path.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java:
##########
@@ -53,6 +54,12 @@ 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>A single vgroup is sufficient for connector assertions and avoids CI
startup failures.
+ */
+ private static final String SINGLE_VGROUP_DATABASE_OPTIONS = " KEEP 3650
VGROUPS 1";
Review Comment:
Confirmed and kept this scoped to CI resource usage. The suite creates and
verifies the STABLE-backed paths `power.meters`, `power2.meters2/3/4`, and
`power3.meters5` by row-count / projection assertions; it does not assert
per-vgroup partitioning. The code comment now explicitly says the single vgroup
is a CI resource workaround and should behave the same for these exercised SQL
paths.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java:
##########
@@ -224,6 +203,37 @@ private Connection createConnect(GenericContainer<?>
tdengineServer) {
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() {
Review Comment:
Addressed in `c21a0d39c1`: `waitForTDengineReady` now creates the fixed
readiness probe database with `CREATE DATABASE IF NOT EXISTS` and intentionally
keeps it until the container stops, instead of dropping and recreating it on
every Awaitility retry. This removes the repeated create/drop churn during
startup.
##########
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 `c21a0d39c1`: expanded
`KafkaITTest.shouldContinueEmptyReadCommittedPollWhenPositionAdvances()` so the
helper explicitly keeps scanning when the consumer position advances across
invisible READ_COMMITTED offsets, even after more than 20 empty polls. This
locks the aborted/control-record boundary in the pure helper logic without
adding broker-heavy E2E work here.
##########
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:
Rechecked this path. `cleanupContinuousJob` preserves the original test
failure by adding cleanup failures as suppressed only when `testFailure` is
non-null, and throws cleanup directly when cleanup is the first failure. I left
the behavior unchanged because it matches the helper contract and avoids
reducing cleanup strictness. If CI shows worst-case timeout pressure, we can
tune the 180s job-exit wait separately with fresh run evidence.
--
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]
