You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2026/07/06 04:23:51 UTC
(flink-cdc) 22/24: [test] Keep stabilization changes semantically strict
This is an automated email from the ASF dual-hosted git repository.
leonardBang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 1357fe1056b25577800990c436f4641b251df7f7
Author: Leonard Xu <xb...@gmail.com>
AuthorDate: Thu Jul 2 23:17:52 2026 +0800
[test] Keep stabilization changes semantically strict
Restore Hudi schema-evolution coverage, keep Mongo snapshot assertions exact, and wait for stream handoff in p=1 cases so the flake fixes do not weaken what these tests prove. Also fail fast when checkpoint triggering targets a missing job instead of treating that as a startup transient.
Co-Authored-By: Claude Opus 4.7 <no...@anthropic.com>
---
.../mongodb/source/NewlyAddedTableITCase.java | 7 +-
.../cdc/pipeline/tests/MySqlToHudiE2eITCase.java | 79 ++++++++++++++++------
.../cdc/pipeline/tests/SqlServerE2eITCase.java | 25 ++++---
.../cdc/pipeline/tests/TransformE2eITCase.java | 31 ++++-----
.../flink/cdc/pipeline/tests/UdfE2eITCase.java | 27 +++++---
.../tests/utils/PipelineTestEnvironment.java | 22 +++++-
6 files changed, 127 insertions(+), 64 deletions(-)
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java
index ac56f4ed6..6bbf75017 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java
@@ -468,11 +468,8 @@ class NewlyAddedTableITCase extends MongoDBSourceTestBase {
format(
"+U[%s, 416874195632735147, China_%s, %s, %s West Town address 1]",
collection0, round, cityName0, cityName0);
- assertEqualsInAnyOrderWithAllowedDuplicateUpdatePair(
- fetchedDataList,
- TestValuesTableFactory.getRawResultsAsStrings("sink"),
- collection0UpdateBefore,
- collection0UpdateAfter);
+ MongoDBAssertUtils.assertEqualsInAnyOrder(
+ fetchedDataList, TestValuesTableFactory.getRawResultsAsStrings("sink"));
// step 4: make changelog data for all collections before this round(also includes this
// round),
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
index 97d849769..45ae95d2a 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
@@ -289,18 +289,39 @@ public class MySqlToHudiE2eITCase extends PipelineTestEnvironment {
"INSERT INTO products VALUES (default,'Ten','Jukebox',0.2, null, null, null);"); // 110
stat.execute("UPDATE products SET description='Fay' WHERE id=106;");
stat.execute("UPDATE products SET weight='5.125' WHERE id=107;");
-
- // Keep this Hudi E2E focused on DML and MOR compaction. Hudi 1.1.0 internal schema
- // commits are flaky on local Docker volumes; schema evolution is covered by other E2Es.
+ stat.execute("ALTER TABLE products DROP COLUMN point_c;");
stat.execute("DELETE FROM products WHERE id=101;");
stat.execute(
- "INSERT INTO products VALUES (default,'Eleven','Kryo',5.18, null, null, null);"); // 111
+ "INSERT INTO products VALUES (default,'Eleven','Kryo',5.18, null, null);"); // 111
+ stat.execute(
+ "INSERT INTO products VALUES (default,'Twelve', 'Lily', 2.14, null, null);"); // 112
+
+ Thread.sleep(2000L);
+ triggerCheckpointOrDescribeFailure(pipelineJobID, "post-drop-column checkpoint");
+ initialProductInstants =
+ waitUntilCompletedHudiInstants(
+ warehouse, database, "products", initialProductInstants + 1);
+ validateSinkResultWithCheckpointProgress(
+ pipelineJobID,
+ "post-drop-column convergence",
+ warehouse,
+ database,
+ "products",
+ getProductsExpectedAfterDropSinkResults());
+
+ stat.execute("ALTER TABLE products ADD COLUMN point_c_0 VARCHAR(10);");
+ stat.execute(
+ "INSERT INTO products VALUES (default,'Thirteen','Mila',3.14, null, null, 'A');"); // 113
+ stat.execute("ALTER TABLE products ADD COLUMN point_c_1 VARCHAR(10);");
+ stat.execute(
+ "INSERT INTO products VALUES (default,'Fourteen','Nora',4.14, null, null, 'B', 'C');"); // 114
+ stat.execute("ALTER TABLE products MODIFY point_c_0 VARCHAR(20);");
stat.execute(
- "INSERT INTO products VALUES (default,'Twelve', 'Lily', 2.14, null, null, null);"); // 112
+ "INSERT INTO products VALUES (default,'Fifteen','Olive',5.14, null, null, 'point-zero-long', 'D');"); // 115
Thread.sleep(2000L);
- triggerCheckpointOrDescribeFailure(pipelineJobID, "post-incremental checkpoint");
+ triggerCheckpointOrDescribeFailure(pipelineJobID, "post-schema-evolution checkpoint");
initialProductInstants =
waitUntilCompletedHudiInstants(
warehouse, database, "products", initialProductInstants + 1);
@@ -311,11 +332,11 @@ public class MySqlToHudiE2eITCase extends PipelineTestEnvironment {
validateSinkResultWithCheckpointProgress(
pipelineJobID,
- "final incremental convergence",
+ "final schema evolution convergence",
warehouse,
database,
"products",
- getProductsExpectedAfterIncrementalSinkResults());
+ getProductsExpectedAfterSchemaEvolutionSinkResults());
if (TABLE_TYPE.equals(HoodieTableType.MERGE_ON_READ.name())) {
waitUntilCompactionScheduled(warehouse, database, "products");
@@ -969,19 +990,37 @@ public class MySqlToHudiE2eITCase extends PipelineTestEnvironment {
"109, Nine, IINA, 5.223, null, null, null");
}
- private static List<String> getProductsExpectedAfterIncrementalSinkResults() {
+ private static List<String> getProductsExpectedAfterDropSinkResults() {
return Arrays.asList(
- "102, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null",
- "103, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null",
- "104, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null",
- "105, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null",
- "106, Six, Fay, 9.813, null, null, null",
- "107, Seven, Grace, 5.125, null, null, null",
- "108, Eight, Hesse, 6.819, null, null, null",
- "109, Nine, IINA, 5.223, null, null, null",
- "110, Ten, Jukebox, 0.2, null, null, null",
- "111, Eleven, Kryo, 5.18, null, null, null",
- "112, Twelve, Lily, 2.14, null, null, null");
+ "102, Two, Bob, 1.703, white, {\"key2\": \"value2\"}",
+ "103, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}",
+ "104, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}",
+ "105, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}",
+ "106, Six, Fay, 9.813, null, null",
+ "107, Seven, Grace, 5.125, null, null",
+ "108, Eight, Hesse, 6.819, null, null",
+ "109, Nine, IINA, 5.223, null, null",
+ "110, Ten, Jukebox, 0.2, null, null",
+ "111, Eleven, Kryo, 5.18, null, null",
+ "112, Twelve, Lily, 2.14, null, null");
+ }
+
+ private static List<String> getProductsExpectedAfterSchemaEvolutionSinkResults() {
+ return Arrays.asList(
+ "102, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null, null",
+ "103, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null, null",
+ "104, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null, null",
+ "105, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null, null",
+ "106, Six, Fay, 9.813, null, null, null, null",
+ "107, Seven, Grace, 5.125, null, null, null, null",
+ "108, Eight, Hesse, 6.819, null, null, null, null",
+ "109, Nine, IINA, 5.223, null, null, null, null",
+ "110, Ten, Jukebox, 0.2, null, null, null, null",
+ "111, Eleven, Kryo, 5.18, null, null, null, null",
+ "112, Twelve, Lily, 2.14, null, null, null, null",
+ "113, Thirteen, Mila, 3.14, null, null, A, null",
+ "114, Fourteen, Nora, 4.14, null, null, B, C",
+ "115, Fifteen, Olive, 5.14, null, null, point-zero-long, D");
}
private static List<String> getCustomersExpectedSinkResults() {
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SqlServerE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SqlServerE2eITCase.java
index 890bcb234..a979a2c64 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SqlServerE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SqlServerE2eITCase.java
@@ -202,18 +202,25 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment {
}
private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
+ Duration readinessTimeout = Duration.ofMinutes(5);
if (parallelism == 1) {
- return;
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished and the job parallelism is 1, snapshot split assigner is turn into finished status.",
+ readinessTimeout);
+ } else {
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.",
+ readinessTimeout);
+ getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner is turn into finished status.",
+ readinessTimeout);
}
-
- waitUntilLogContains(
- jobManagerConsumer,
- "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.");
- getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
waitUntilLogContains(
- jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
- waitUntilLogContains(jobManagerConsumer, "Assign split StreamSplit{splitId='stream-split'");
- waitUntilLogContains(jobManagerConsumer, "for the stream split assignment.");
+ jobManagerConsumer, "for the stream split assignment.", readinessTimeout);
}
private void waitUntilJobRunningFromClusterCli(JobID jobId, Duration timeout) throws Exception {
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
index ef3105b87..bdbdcd9b8 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
@@ -1322,24 +1322,23 @@ class TransformE2eITCase extends PipelineTestEnvironment {
}
private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
+ Duration readinessTimeout = Duration.ofMinutes(5);
if (parallelism == 1) {
- return;
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished and the job parallelism is 1, snapshot split assigner is turn into finished status.",
+ readinessTimeout);
+ } else {
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.",
+ readinessTimeout);
+ getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner is turn into finished status.",
+ readinessTimeout);
}
-
- Duration readinessTimeout = Duration.ofMinutes(5);
- waitUntilLogContains(
- jobManagerConsumer,
- "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.",
- readinessTimeout);
- getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
- waitUntilLogContains(
- jobManagerConsumer,
- "Snapshot split assigner is turn into finished status.",
- readinessTimeout);
- waitUntilLogContains(
- jobManagerConsumer,
- "The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'",
- readinessTimeout);
waitUntilLogContains(
jobManagerConsumer, "for the binlog split assignment.", readinessTimeout);
}
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java
index 147525560..9ef91b30b 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java
@@ -286,19 +286,24 @@ class UdfE2eITCase extends PipelineTestEnvironment {
}
private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
+ Duration readinessTimeout = Duration.ofMinutes(5);
if (parallelism == 1) {
- return;
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished and the job parallelism is 1, snapshot split assigner is turn into finished status.",
+ readinessTimeout);
+ } else {
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.",
+ readinessTimeout);
+ getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
+ waitUntilLogContains(
+ jobManagerConsumer,
+ "Snapshot split assigner is turn into finished status.",
+ readinessTimeout);
}
-
- waitUntilLogContains(
- jobManagerConsumer,
- "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.");
- getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
- waitUntilLogContains(
- jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
waitUntilLogContains(
- jobManagerConsumer,
- "The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'");
- waitUntilLogContains(jobManagerConsumer, "for the binlog split assignment.");
+ jobManagerConsumer, "for the binlog split assignment.", readinessTimeout);
}
}
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
index 4d4187801..1a49f1c39 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
@@ -358,15 +358,22 @@ public abstract class PipelineTestEnvironment extends TestLogger {
Optional<FlinkJobNotFoundException> jobNotFoundException =
ExceptionUtils.findThrowable(e, FlinkJobNotFoundException.class);
String errorMessage = ExceptionUtils.stringifyException(e);
+ if (jobNotFoundException.isPresent()
+ || errorMessage.contains("Could not find Flink job")) {
+ throw new IllegalStateException(
+ "Failed to trigger checkpoint for job "
+ + jobID
+ + " because the job was not found. "
+ + describeCheckpointTarget(jobID),
+ e);
+ }
if ((checkpointException.isPresent()
&& checkpointException
.get()
.getMessage()
.contains("Checkpoint triggering task"))
|| errorMessage.contains("is not being executed at the moment")
- || errorMessage.contains("Not all required tasks are currently running")
- || errorMessage.contains("Could not find Flink job")
- || jobNotFoundException.isPresent()) {
+ || errorMessage.contains("Not all required tasks are currently running")) {
Thread.sleep(100L);
retryTimes++;
} else {
@@ -377,6 +384,15 @@ public abstract class PipelineTestEnvironment extends TestLogger {
throw new TimeoutException("Timed out waiting to trigger checkpoint for job " + jobID);
}
+ private String describeCheckpointTarget(JobID jobID) {
+ try {
+ JobStatus status = getRestClusterClient().getJobStatus(jobID).get(10, TimeUnit.SECONDS);
+ return "Current job status: " + status + '.';
+ } catch (Exception statusError) {
+ return "Job status is unavailable: " + ExceptionUtils.stringifyException(statusError);
+ }
+ }
+
/**
* Get {@link RestClusterClient} connected to this FlinkContainer.
*