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 7f90244acb1671c74bcbebff045bbd43b9e81122 Author: Leonard Xu <[email protected]> AuthorDate: Wed Jul 1 14:57:47 2026 +0800 [test][pipeline-e2e] Stabilize SqlServerE2eITCase split handoff --- .../cdc/pipeline/tests/SqlServerE2eITCase.java | 55 ++++++++++++++++++++-- 1 file changed, 50 insertions(+), 5 deletions(-) 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 c961206ae..890bcb234 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 @@ -17,8 +17,11 @@ 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; @@ -26,6 +29,7 @@ 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.sql.Statement; 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 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { 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 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { 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 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { 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,33 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { SQL_SERVER_CONTAINER.getUsername(), SQL_SERVER_CONTAINER.getPassword()); } + + private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { + if (parallelism == 1) { + return; + } + + 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."); + } + + 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)); + } }
