Skip to content

Commit 907ffd3

Browse files
leonardBangclaude
leonardBang
and
claude
committed
[test][pipeline-e2e] Improve MySQL pipeline test stability before incremental reads
TransformE2eITCase and UdfE2eITCase could generate binlog changes before the stream split was assigned under multi-parallelism, racing snapshot completion and producing flaky incremental assertions. Capture the job ID, trigger a checkpoint-gated readiness barrier, and wait until the binlog split is assigned before writing incremental changes. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1 parent 11f520c commit 907ffd3

2 files changed

Lines changed: 69 additions & 15 deletions

File tree

‎flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java‎

Lines changed: 44 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,14 @@
1717

1818
package org.apache.flink.cdc.pipeline.tests;
1919

20+
import org.apache.flink.api.common.JobID;
2021
import org.apache.flink.cdc.common.data.DateData;
2122
import org.apache.flink.cdc.common.data.TimeData;
2223
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
2324
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
2425
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
2526
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
27+
import org.apache.flink.core.execution.CheckpointType;
2628

2729
import org.junit.jupiter.api.AfterEach;
2830
import org.junit.jupiter.api.BeforeEach;
@@ -117,7 +119,7 @@ void testHeteroSchemaTransform(boolean batchMode) throws Exception {
117119
transformTestDatabase.getDatabaseName(),
118120
runtimeMode,
119121
parallelism);
120-
submitPipelineJob(pipelineJob);
122+
JobID jobId = submitPipelineJob(pipelineJob);
121123
waitUntilJobRunning(Duration.ofSeconds(30));
122124
LOG.info("Pipeline job is running");
123125

@@ -137,6 +139,7 @@ void testHeteroSchemaTransform(boolean batchMode) throws Exception {
137139
}
138140

139141
LOG.info("Begin incremental reading stage.");
142+
waitUntilStreamSplitReady(jobId, parallelism);
140143
// generate binlogs
141144
String mysqlJdbcUrl =
142145
String.format(
@@ -199,7 +202,7 @@ void testMultipleTransformRule(boolean batchMode) throws Exception {
199202
transformTestDatabase.getDatabaseName(),
200203
runtimeMode,
201204
parallelism);
202-
submitPipelineJob(pipelineJob);
205+
JobID jobId = submitPipelineJob(pipelineJob);
203206
waitUntilJobRunning(Duration.ofSeconds(30));
204207
LOG.info("Pipeline job is running");
205208

@@ -221,6 +224,7 @@ void testMultipleTransformRule(boolean batchMode) throws Exception {
221224
}
222225

223226
LOG.info("Begin incremental reading stage.");
227+
waitUntilStreamSplitReady(jobId, parallelism);
224228
// generate binlogs
225229
String mysqlJdbcUrl =
226230
String.format(
@@ -289,7 +293,7 @@ void testAssortedSchemaTransform(boolean batchMode) throws Exception {
289293
transformTestDatabase.getDatabaseName(),
290294
runtimeMode,
291295
parallelism);
292-
submitPipelineJob(pipelineJob);
296+
JobID jobId = submitPipelineJob(pipelineJob);
293297
waitUntilJobRunning(Duration.ofSeconds(30));
294298
LOG.info("Pipeline job is running");
295299

@@ -308,6 +312,7 @@ void testAssortedSchemaTransform(boolean batchMode) throws Exception {
308312
}
309313

310314
LOG.info("Begin incremental reading stage.");
315+
waitUntilStreamSplitReady(jobId, parallelism);
311316
// generate binlogs
312317
String mysqlJdbcUrl =
313318
String.format(
@@ -371,7 +376,7 @@ void testWildcardSchemaTransform(boolean batchMode) throws Exception {
371376
transformTestDatabase.getDatabaseName(),
372377
runtimeMode,
373378
parallelism);
374-
submitPipelineJob(pipelineJob);
379+
JobID jobId = submitPipelineJob(pipelineJob);
375380
waitUntilJobRunning(Duration.ofSeconds(30));
376381
LOG.info("Pipeline job is running");
377382

@@ -391,6 +396,7 @@ void testWildcardSchemaTransform(boolean batchMode) throws Exception {
391396
}
392397

393398
LOG.info("Begin incremental reading stage.");
399+
waitUntilStreamSplitReady(jobId, parallelism);
394400
// generate binlogs
395401
String mysqlJdbcUrl =
396402
String.format(
@@ -454,7 +460,7 @@ void testWildcardWithMetadataColumnTransform(boolean batchMode) throws Exception
454460
transformTestDatabase.getDatabaseName(),
455461
runtimeMode,
456462
parallelism);
457-
submitPipelineJob(pipelineJob);
463+
JobID jobId = submitPipelineJob(pipelineJob);
458464
waitUntilJobRunning(Duration.ofSeconds(30));
459465
LOG.info("Pipeline job is running");
460466

@@ -475,6 +481,7 @@ void testWildcardWithMetadataColumnTransform(boolean batchMode) throws Exception
475481
return;
476482
}
477483

484+
waitUntilStreamSplitReady(jobId, parallelism);
478485
// generate binlogs
479486
String mysqlJdbcUrl =
480487
String.format(
@@ -537,7 +544,7 @@ void testMultipleHittingTable(boolean batchMode) throws Exception {
537544
transformTestDatabase.getDatabaseName(),
538545
runtimeMode,
539546
parallelism);
540-
submitPipelineJob(pipelineJob);
547+
JobID jobId = submitPipelineJob(pipelineJob);
541548
waitUntilJobRunning(Duration.ofSeconds(30));
542549
LOG.info("Pipeline job is running");
543550

@@ -558,6 +565,7 @@ void testMultipleHittingTable(boolean batchMode) throws Exception {
558565
return;
559566
}
560567

568+
waitUntilStreamSplitReady(jobId, parallelism);
561569
// generate binlogs
562570
String mysqlJdbcUrl =
563571
String.format(
@@ -617,7 +625,7 @@ void testMultipleTransformWithDiffRefColumn(boolean batchMode) throws Exception
617625
transformTestDatabase.getDatabaseName(),
618626
runtimeMode,
619627
parallelism);
620-
submitPipelineJob(pipelineJob);
628+
JobID jobId = submitPipelineJob(pipelineJob);
621629
waitUntilJobRunning(Duration.ofSeconds(30));
622630
LOG.info("Pipeline job is running");
623631

@@ -633,6 +641,7 @@ void testMultipleTransformWithDiffRefColumn(boolean batchMode) throws Exception
633641
return;
634642
}
635643

644+
waitUntilStreamSplitReady(jobId, parallelism);
636645
// generate binlogs
637646
String mysqlJdbcUrl =
638647
String.format(
@@ -696,7 +705,7 @@ void testTransformWithCast(boolean batchMode) throws Exception {
696705
transformTestDatabase.getDatabaseName(),
697706
runtimeMode,
698707
parallelism);
699-
submitPipelineJob(pipelineJob);
708+
JobID jobId = submitPipelineJob(pipelineJob);
700709
waitUntilJobRunning(Duration.ofSeconds(30));
701710

702711
if (batchMode) {
@@ -715,6 +724,7 @@ void testTransformWithCast(boolean batchMode) throws Exception {
715724
"DataChangeEvent{tableId=%s.TABLEBETA, before=[], after=[2014, 114.0, 24 - Henry], op=INSERT, meta=()}");
716725

717726
LOG.info("Begin incremental reading stage.");
727+
waitUntilStreamSplitReady(jobId, parallelism);
718728
// generate binlogs
719729
String mysqlJdbcUrl =
720730
String.format(
@@ -777,7 +787,7 @@ void testTemporalFunctions(boolean batchMode) throws Exception {
777787
transformTestDatabase.getDatabaseName(),
778788
runtimeMode,
779789
parallelism);
780-
submitPipelineJob(pipelineJob);
790+
JobID jobId = submitPipelineJob(pipelineJob);
781791
waitUntilJobRunning(Duration.ofSeconds(30));
782792
LOG.info("Pipeline job is running");
783793

@@ -812,7 +822,7 @@ void testTransformWithSchemaEvolution() throws Exception {
812822
transformTestDatabase.getDatabaseName(),
813823
transformTestDatabase.getDatabaseName(),
814824
parallelism);
815-
submitPipelineJob(pipelineJob);
825+
JobID jobId = submitPipelineJob(pipelineJob);
816826
waitUntilJobRunning(Duration.ofSeconds(30));
817827
LOG.info("Pipeline job is running");
818828

@@ -824,6 +834,7 @@ void testTransformWithSchemaEvolution() throws Exception {
824834
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011, id -> 1011, 59], op=INSERT, meta=()}");
825835

826836
LOG.info("Begin incremental reading stage.");
837+
waitUntilStreamSplitReady(jobId, parallelism);
827838
// generate binlogs
828839
String mysqlJdbcUrl =
829840
String.format(
@@ -906,7 +917,7 @@ void testTransformWildcardPrefixWithSchemaEvolution() throws Exception {
906917
transformTestDatabase.getDatabaseName(),
907918
transformTestDatabase.getDatabaseName(),
908919
parallelism);
909-
submitPipelineJob(pipelineJob);
920+
JobID jobId = submitPipelineJob(pipelineJob);
910921
waitUntilJobRunning(Duration.ofSeconds(30));
911922
LOG.info("Pipeline job is running");
912923

@@ -918,6 +929,7 @@ void testTransformWildcardPrefixWithSchemaEvolution() throws Exception {
918929
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011, 11, 59, 20, Dave, id -> 1011], op=INSERT, meta=()}");
919930

920931
LOG.info("Begin incremental reading stage.");
932+
waitUntilStreamSplitReady(jobId, parallelism);
921933
// generate binlogs
922934
String mysqlJdbcUrl =
923935
String.format(
@@ -1008,7 +1020,7 @@ void testTransformWildcardSuffixWithSchemaEvolution() throws Exception {
10081020
transformTestDatabase.getDatabaseName(),
10091021
transformTestDatabase.getDatabaseName(),
10101022
parallelism);
1011-
submitPipelineJob(pipelineJob);
1023+
JobID jobId = submitPipelineJob(pipelineJob);
10121024
waitUntilJobRunning(Duration.ofSeconds(30));
10131025
LOG.info("Pipeline job is running");
10141026

@@ -1020,6 +1032,7 @@ void testTransformWildcardSuffixWithSchemaEvolution() throws Exception {
10201032
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011 <- id, 1011, 11, 59, 20, Dave], op=INSERT, meta=()}");
10211033

10221034
LOG.info("Begin incremental reading stage.");
1035+
waitUntilStreamSplitReady(jobId, parallelism);
10231036
// generate binlogs
10241037
String mysqlJdbcUrl =
10251038
String.format(
@@ -1140,7 +1153,7 @@ void testTransformWithUnicodeLiterals() throws Exception {
11401153
transformTestDatabase.getDatabaseName(),
11411154
projectionExpression,
11421155
parallelism);
1143-
submitPipelineJob(pipelineJob);
1156+
JobID jobId = submitPipelineJob(pipelineJob);
11441157
waitUntilJobRunning(Duration.ofSeconds(30));
11451158
LOG.info("Pipeline job is running");
11461159

@@ -1152,6 +1165,7 @@ void testTransformWithUnicodeLiterals() throws Exception {
11521165
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[1011, 11, 59, 20, Dave, ascii test!?, 大五, 测试数据, ひびぴ, 죠주쥬, ÀÆÉ, ÓÔŐÖ, αβγδε, בבקשה, твой, ภาษาไทย, piedzimst brīvi], op=INSERT, meta=()}");
11531166

11541167
LOG.info("Begin incremental reading stage.");
1168+
waitUntilStreamSplitReady(jobId, parallelism);
11551169
// generate binlogs
11561170
String mysqlJdbcUrl =
11571171
String.format(
@@ -1216,6 +1230,23 @@ void testTransformWithUnicodeLiterals() throws Exception {
12161230
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[Beginning, 3010, 10, 10, 97, Lemon, ascii test!?, 大五, 测试数据, ひびぴ, 죠주쥬, ÀÆÉ, ÓÔŐÖ, αβγδε, בבקשה, твой, ภาษาไทย, piedzimst brīvi], op=INSERT, meta=()}");
12171231
}
12181232

1233+
private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
1234+
if (parallelism == 1) {
1235+
return;
1236+
}
1237+
1238+
waitUntilSpecificEvent(
1239+
jobManagerConsumer,
1240+
"Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.");
1241+
getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
1242+
waitUntilSpecificEvent(
1243+
jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
1244+
waitUntilSpecificEvent(
1245+
jobManagerConsumer,
1246+
"The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'");
1247+
waitUntilSpecificEvent(jobManagerConsumer, "for the binlog split assignment.");
1248+
}
1249+
12191250
private void validateEventsWithPattern(String... patterns) throws Exception {
12201251
for (String pattern : patterns) {
12211252
waitUntilSpecificEventWithPattern(

‎flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,11 @@
1717

1818
package org.apache.flink.cdc.pipeline.tests;
1919

20+
import org.apache.flink.api.common.JobID;
2021
import org.apache.flink.cdc.common.test.utils.TestUtils;
2122
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
2223
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
24+
import org.apache.flink.core.execution.CheckpointType;
2325

2426
import org.junit.jupiter.api.AfterEach;
2527
import org.junit.jupiter.api.BeforeEach;
@@ -125,7 +127,7 @@ void testUserDefinedFunctions(String language, boolean batchMode) throws Excepti
125127
language);
126128
Path udfJar = TestUtils.getResource("udf-examples.jar");
127129
Path scalaLibJar = TestUtils.getResource("scala-library.jar");
128-
submitPipelineJob(pipelineJob, udfJar, scalaLibJar);
130+
JobID jobId = submitPipelineJob(pipelineJob, udfJar, scalaLibJar);
129131
waitUntilJobRunning(Duration.ofSeconds(30));
130132
waitUntilSpecificEvent("[ LifecycleFunction ] opened.");
131133

@@ -145,6 +147,8 @@ void testUserDefinedFunctions(String language, boolean batchMode) throws Excepti
145147
return;
146148
}
147149

150+
waitUntilStreamSplitReady(jobId, parallelism);
151+
148152
String mysqlJdbcUrl =
149153
String.format(
150154
"jdbc:mysql://%s:%s/%s",
@@ -221,7 +225,7 @@ void testFlinkCompatibleScalarFunctions(String language, boolean batchMode) thro
221225
language);
222226
Path udfJar = TestUtils.getResource("udf-examples.jar");
223227
Path scalaLibJar = TestUtils.getResource("scala-library.jar");
224-
submitPipelineJob(pipelineJob, udfJar, scalaLibJar);
228+
JobID jobId = submitPipelineJob(pipelineJob, udfJar, scalaLibJar);
225229
waitUntilJobRunning(Duration.ofSeconds(30));
226230
validateResult(
227231
dbNameFormatter,
@@ -239,6 +243,8 @@ void testFlinkCompatibleScalarFunctions(String language, boolean batchMode) thro
239243
return;
240244
}
241245

246+
waitUntilStreamSplitReady(jobId, parallelism);
247+
242248
String mysqlJdbcUrl =
243249
String.format(
244250
"jdbc:mysql://%s:%s/%s",
@@ -263,4 +269,21 @@ void testFlinkCompatibleScalarFunctions(String language, boolean batchMode) thro
263269
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[3007, 7, 3009, <7>], op=INSERT, meta=()}",
264270
"DataChangeEvent{tableId=%s.TABLEBETA, before=[2011, 11, Integer: 2011], after=[], op=DELETE, meta=()}");
265271
}
272+
273+
private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
274+
if (parallelism == 1) {
275+
return;
276+
}
277+
278+
waitUntilSpecificEvent(
279+
jobManagerConsumer,
280+
"Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.");
281+
getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
282+
waitUntilSpecificEvent(
283+
jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
284+
waitUntilSpecificEvent(
285+
jobManagerConsumer,
286+
"The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'");
287+
waitUntilSpecificEvent(jobManagerConsumer, "for the binlog split assignment.");
288+
}
266289
}

0 commit comments

Comments
 (0)