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 <[email protected]> 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 <[email protected]> --- .../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. *
