[test][pipeline-e2e] Improve SqlServerE2eITCase startup and stream split stability

SqlServerE2eITCase could proceed before the job was fully visible to the
cluster CLI or before the stream split was assigned, making the first
incremental assertions race startup. Wait for the submitted job to appear in
`flink list`, gate the snapshot-to-stream transition on a completed
checkpoint, and split the initial INSERT from later update/delete assertions.

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/SqlServerE2eITCase.java

@@ -17,15 +17,19 @@
 
 package org.apache.flink.cdc.pipeline.tests;
 
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Deadline;
 import org.apache.flink.cdc.common.test.utils.TestUtils;
 import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+import org.apache.flink.core.execution.CheckpointType;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container.ExecResult;
 import org.testcontainers.containers.MSSQLServerContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.junit.jupiter.Container;
@@ -42,6 +46,8 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -108,8 +114,8 @@ void testSyncWholeDatabase() throws Exception {
                         SQL_SERVER_CONTAINER.getPassword(),
                         parallelism);
         Path sqlServerCdcJar = TestUtils.getResource("sqlserver-cdc-pipeline-connector.jar");
-        submitPipelineJob(pipelineJob, sqlServerCdcJar);
-        waitUntilJobRunning(Duration.ofSeconds(30));
+        JobID jobId = submitPipelineJob(pipelineJob, sqlServerCdcJar);
+        waitUntilJobRunningFromClusterCli(jobId, Duration.ofSeconds(30));
         LOG.info("Pipeline job is running");
 
         validateResult(
@@ -126,11 +132,24 @@ void testSyncWholeDatabase() throws Exception {
 
         LOG.info("Begin incremental reading stage.");
 
+        waitUntilStreamSplitReady(jobId, parallelism);
+
         try (Connection conn = getSqlServerJdbcConnection();
                 Statement stat = conn.createStatement()) {
             stat.execute("USE inventory;");
             stat.execute(
                     "INSERT INTO dbo.products(id,name,description,weight) VALUES (110,'jacket','water resistent white wind breaker',0.2);");
+        } catch (SQLException e) {
+            LOG.error("Insert row for CDC failed.", e);
+            throw e;
+        }
+
+        waitUntilSpecificEvent(
+                "DataChangeEvent{tableId=inventory.dbo.products, before=[], after=[110, jacket, water resistent white wind breaker, 0.2], op=INSERT, meta=()}");
+
+        try (Connection conn = getSqlServerJdbcConnection();
+                Statement stat = conn.createStatement()) {
+            stat.execute("USE inventory;");
             stat.execute(
                     "UPDATE dbo.products SET description='18oz carpenter hammer' WHERE id=106;");
             stat.execute("UPDATE dbo.products SET weight=5.1 WHERE id=107;");
@@ -139,9 +158,6 @@ void testSyncWholeDatabase() throws Exception {
             LOG.error("Update table for CDC failed.", e);
             throw e;
         }
-
-        waitUntilSpecificEvent(
-                "DataChangeEvent{tableId=inventory.dbo.products, before=[], after=[110, jacket, water resistent white wind breaker, 0.2], op=INSERT, meta=()}");
         waitUntilSpecificEvent(
                 "DataChangeEvent{tableId=inventory.dbo.products, before=[106, hammer, 16oz carpenter's hammer, 1.0], after=[106, hammer, 18oz carpenter hammer, 1.0], op=UPDATE, meta=()}");
         waitUntilSpecificEvent(
@@ -184,4 +200,36 @@ private Connection getSqlServerJdbcConnection() throws SQLException {
                 SQL_SERVER_CONTAINER.getUsername(),
                 SQL_SERVER_CONTAINER.getPassword());
     }
+
+    private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception {
+        if (parallelism == 1) {
+            return;
+        }
+
+        waitUntilSpecificEvent(
+                jobManagerConsumer,
+                "Snapshot split assigner received all splits finished, waiting for a complete checkpoint to mark the assigner finished.");
+        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'");
+    }
+
+    private void waitUntilJobRunningFromClusterCli(JobID jobId, Duration timeout) throws Exception {
+        Deadline deadline = Deadline.fromNow(timeout);
+        String jobIdHex = jobId.toHexString();
+        while (deadline.hasTimeLeft()) {
+            ExecResult execResult = jobManager.execInContainer("bash", "-lc", "flink list");
+            if (execResult.getExitCode() == 0 && execResult.getStdout().contains(jobIdHex)) {
+                return;
+            }
+            TimeUnit.SECONDS.sleep(1);
+        }
+        throw new TimeoutException(
+                String.format("Timed out waiting for job %s to appear in `flink list`.", jobIdHex));
+    }
 }