[test][pipeline-e2e] Improve split wait matching and Iceberg checkpoint stability

Use substring-based split readiness waits so parallel pipeline tests stop blocking on exact log lines that never appear. Trigger a final checkpoint before the MySqlToIceberg end-to-end validation so the last incremental batch is committed before asserting full table contents.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: leonardBang

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

@@ -17,10 +17,12 @@
 
 package org.apache.flink.cdc.pipeline.tests;
 
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.cdc.common.test.utils.TestUtils;
 import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
 import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
 import org.apache.flink.cdc.pipeline.tests.utils.TarballFetcher;
+import org.apache.flink.core.execution.CheckpointType;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.CatalogUtil;
@@ -168,7 +170,9 @@ public void testSyncWholeDatabase() throws Exception {
         Path icebergCdcConnector = TestUtils.getResource("iceberg-cdc-pipeline-connector.jar");
         Path hadoopJar = TestUtils.getResource("flink-shade-hadoop.jar");
         Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
-        submitPipelineJob(pipelineJob, mysqlCdcJar, icebergCdcConnector, mysqlDriverJar, hadoopJar);
+        JobID jobId =
+                submitPipelineJob(
+                        pipelineJob, mysqlCdcJar, icebergCdcConnector, mysqlDriverJar, hadoopJar);
         waitUntilJobRunning(Duration.ofSeconds(60));
         LOG.info("Pipeline job is running");
         validateSinkResult(
@@ -254,6 +258,7 @@ public void testSyncWholeDatabase() throws Exception {
                             "112, Twelve, Lily, 2.14, null, null"));
 
             recordsInIncrementalPhase = createChangesAndValidate(stat);
+            getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
         } catch (SQLException e) {
             LOG.error("Update table for CDC failed.", e);
             throw e;

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

@@ -212,11 +212,8 @@ private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exce
         getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
         waitUntilSpecificEvent(
                 jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
-        waitUntilSpecificEvent(
-                jobManagerConsumer, "Assign split StreamSplit{splitId='stream-split'");
-        waitUntilSpecificEvent(
-                taskManagerConsumer,
-                "Adding split(s) to reader: [StreamSplit{splitId='stream-split'");
+        waitUntilLogContains(jobManagerConsumer, "Assign split StreamSplit{splitId='stream-split'");
+        waitUntilLogContains(jobManagerConsumer, "for the stream split assignment.");
     }
 
     private void waitUntilJobRunningFromClusterCli(JobID jobId, Duration timeout) throws Exception {

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

@@ -1241,10 +1241,10 @@ private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exce
         getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
         waitUntilSpecificEvent(
                 jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
-        waitUntilSpecificEvent(
+        waitUntilLogContains(
                 jobManagerConsumer,
                 "The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'");
-        waitUntilSpecificEvent(jobManagerConsumer, "for the binlog split assignment.");
+        waitUntilLogContains(jobManagerConsumer, "for the binlog split assignment.");
     }
 
     private void validateEventsWithPattern(String... patterns) throws Exception {

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

@@ -281,9 +281,9 @@ private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exce
         getRestClusterClient().triggerCheckpoint(jobId, CheckpointType.CONFIGURED).get();
         waitUntilSpecificEvent(
                 jobManagerConsumer, "Snapshot split assigner is turn into finished status.");
-        waitUntilSpecificEvent(
+        waitUntilLogContains(
                 jobManagerConsumer,
                 "The enumerator assigns split MySqlBinlogSplit{splitId='binlog-split'");
-        waitUntilSpecificEvent(jobManagerConsumer, "for the binlog split assignment.");
+        waitUntilLogContains(jobManagerConsumer, "for the binlog split assignment.");
     }
 }

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java

@@ -510,4 +510,24 @@ protected void waitUntilSpecificEvent(ToStringConsumer consumer, String event)
                             + consumer.toUtf8String());
         }
     }
+
+    protected void waitUntilLogContains(ToStringConsumer consumer, String fragment)
+            throws Exception {
+        boolean result = false;
+        long endTimeout = System.currentTimeMillis() + EVENT_WAITING_TIMEOUT.toMillis();
+        while (System.currentTimeMillis() < endTimeout) {
+            if (consumer.toUtf8String().contains(fragment)) {
+                result = true;
+                break;
+            }
+            Thread.sleep(1000);
+        }
+        if (!result) {
+            throw new TimeoutException(
+                    "failed to get log fragment: "
+                            + fragment
+                            + " from stdout: "
+                            + consumer.toUtf8String());
+        }
+    }
 }