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 accee655242e1366558f04ddba206aa6e3efe69d Author: Leonard Xu <[email protected]> AuthorDate: Sun Jul 5 21:52:59 2026 +0800 [test][pipeline-e2e] Share stream split readiness wait --- .../cdc/pipeline/tests/MysqlToKafkaE2eITCase.java | 23 ------------------- .../cdc/pipeline/tests/SqlServerE2eITCase.java | 25 +-------------------- .../cdc/pipeline/tests/TransformE2eITCase.java | 23 ------------------- .../flink/cdc/pipeline/tests/UdfE2eITCase.java | 23 ------------------- .../tests/utils/PipelineTestEnvironment.java | 26 ++++++++++++++++++++++ 5 files changed, 27 insertions(+), 93 deletions(-) diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlToKafkaE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlToKafkaE2eITCase.java index fe8d5c0fb..f02ba986b 100644 --- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlToKafkaE2eITCase.java +++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlToKafkaE2eITCase.java @@ -23,7 +23,6 @@ import org.apache.flink.cdc.common.test.utils.TestUtils; import org.apache.flink.cdc.connectors.kafka.sink.KafkaUtil; import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase; import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment; -import org.apache.flink.core.execution.CheckpointType; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; @@ -352,28 +351,6 @@ class MysqlToKafkaE2eITCase extends PipelineTestEnvironment { .containsExactlyInAnyOrderElementsOf(deserializeValues(collectedRecords)); } - private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { - Duration readinessTimeout = Duration.ofMinutes(5); - if (parallelism == 1) { - 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, "for the binlog split assignment.", readinessTimeout); - } - private void waitUntilSpecificEventCount( List<ConsumerRecord<byte[], byte[]>> actualEvent, int expectedCount) throws Exception { boolean result = false; 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 a979a2c64..ad24ff275 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 @@ -21,7 +21,6 @@ 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; @@ -132,7 +131,7 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { LOG.info("Begin incremental reading stage."); - waitUntilStreamSplitReady(jobId, parallelism); + waitUntilStreamSplitReady(jobId, parallelism, "for the stream split assignment."); try (Connection conn = getSqlServerJdbcConnection(); Statement stat = conn.createStatement()) { @@ -201,28 +200,6 @@ public class SqlServerE2eITCase extends PipelineTestEnvironment { SQL_SERVER_CONTAINER.getPassword()); } - private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { - Duration readinessTimeout = Duration.ofMinutes(5); - if (parallelism == 1) { - 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, "for the stream split assignment.", readinessTimeout); - } - private void waitUntilJobRunningFromClusterCli(JobID jobId, Duration timeout) throws Exception { Deadline deadline = Deadline.fromNow(timeout); String jobIdHex = jobId.toHexString(); 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 bdbdcd9b8..4f9c7bd19 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 @@ -24,7 +24,6 @@ import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase; import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment; import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator; import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator; -import org.apache.flink.core.execution.CheckpointType; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -1321,28 +1320,6 @@ class TransformE2eITCase extends PipelineTestEnvironment { "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[Beginning, 3010, 10, 10, 97, Lemon, ascii test!?, 大五, 测试数据, ひびぴ, 죠주쥬, ÀÆÉ, ÓÔŐÖ, αβγδε, בבקשה, твой, ภาษาไทย, piedzimst brīvi], op=INSERT, meta=()}"); } - private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { - Duration readinessTimeout = Duration.ofMinutes(5); - if (parallelism == 1) { - 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, "for the binlog split assignment.", readinessTimeout); - } - private void validateEventsWithPattern(String... patterns) throws Exception { for (String pattern : patterns) { waitUntilSpecificEventWithPattern( 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 9ef91b30b..24b76f721 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 @@ -21,7 +21,6 @@ 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.core.execution.CheckpointType; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -284,26 +283,4 @@ class UdfE2eITCase extends PipelineTestEnvironment { "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[3007, 7, 3009, <7>], op=INSERT, meta=()}", "DataChangeEvent{tableId=%s.TABLEBETA, before=[2011, 11, Integer: 2011], after=[], op=DELETE, meta=()}"); } - - private void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { - Duration readinessTimeout = Duration.ofMinutes(5); - if (parallelism == 1) { - 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, "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 1a49f1c39..65e8448f1 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 @@ -384,6 +384,32 @@ public abstract class PipelineTestEnvironment extends TestLogger { throw new TimeoutException("Timed out waiting to trigger checkpoint for job " + jobID); } + protected void waitUntilStreamSplitReady(JobID jobId, int parallelism) throws Exception { + waitUntilStreamSplitReady(jobId, parallelism, "for the binlog split assignment."); + } + + protected void waitUntilStreamSplitReady( + JobID jobId, int parallelism, String streamSplitAssignmentLog) throws Exception { + Duration readinessTimeout = Duration.ofMinutes(5); + if (parallelism == 1) { + 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); + triggerCheckpointWithRetry(jobId); + waitUntilLogContains( + jobManagerConsumer, + "Snapshot split assigner is turn into finished status.", + readinessTimeout); + } + waitUntilLogContains(jobManagerConsumer, streamSplitAssignmentLog, readinessTimeout); + } + private String describeCheckpointTarget(JobID jobID) { try { JobStatus status = getRestClusterClient().getJobStatus(jobID).get(10, TimeUnit.SECONDS);
