This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 59790212e0 [tests][ci] Split pipeline e2e into stages to avoid CI
timeout (#4546)
59790212e0 is described below
commit 59790212e0b9402a48f8293be17b9f680eecb39b
Author: yuxiqian <[email protected]>
AuthorDate: Thu Sep 17 11:14:11 2026 +0800
[tests][ci] Split pipeline e2e into stages to avoid CI timeout (#4546)
---
.github/workflows/flink_cdc_base.yml | 52 +++++++++++++++++-----
.github/workflows/flink_cdc_ci.yml | 18 ++------
.github/workflows/flink_cdc_ci_nightly.yml | 18 ++------
.../flink-cdc-pipeline-e2e-tests/pom.xml | 11 ++++-
.../tests/{ => stage1}/MysqlE2eITCase.java | 10 ++++-
.../MysqlE2eWithYarnApplicationITCase.java | 2 +-
.../pipeline/tests/{ => stage1}/UdfE2eITCase.java | 12 ++++-
.../tests/{ => stage1}/ValuesE2eITCase.java | 10 ++++-
.../migration/MySqlToPaimonMigrationITCase.java | 11 ++++-
.../migration/YamlJobMigrationITCase.java | 10 ++++-
.../tests/{ => stage1}/specs/CheckStep.java | 2 +-
.../tests/{ => stage1}/specs/ExecStep.java | 2 +-
.../specs/FlinkPipelineSpecsITCase.java | 10 ++++-
.../tests/{ => stage1}/specs/RuleSpec.java | 2 +-
.../tests/{ => stage1}/specs/SpecStep.java | 2 +-
.../tests/{ => stage1}/specs/SubmitStep.java | 2 +-
.../tests/{ => stage2}/AiFunctionE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/FlussE2eITCase.java | 9 +++-
.../tests/{ => stage2}/MySqlToDorisE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/MySqlToHudiE2eITCase.java | 10 ++++-
.../{ => stage2}/MySqlToIcebergE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/MySqlToPaimonE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/MysqlToKafkaE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/PythonUdfE2eITCase.java | 10 ++++-
.../tests/{ => stage2}/SqlServerE2eITCase.java | 10 ++++-
.../tests/{ => stage3}/MaxComputeE2eITCase.java | 10 ++++-
.../tests/{ => stage3}/OceanBaseE2eITCase.java | 10 ++++-
.../tests/{ => stage3}/OracleE2eITCase.java | 11 ++++-
.../tests/{ => stage3}/PostgresE2eITCase.java | 10 ++++-
.../tests/{ => stage3}/RouteE2eITCase.java | 9 +++-
.../tests/{ => stage3}/SchemaEvolveE2eITCase.java | 10 ++++-
.../SchemaEvolvingTransformE2eITCase.java | 10 ++++-
.../tests/{ => stage3}/TransformE2eITCase.java | 9 +++-
.../tests/utils/PipelineTestEnvironment.java | 13 ++----
34 files changed, 275 insertions(+), 80 deletions(-)
diff --git a/.github/workflows/flink_cdc_base.yml
b/.github/workflows/flink_cdc_base.yml
index c71195e171..e805dc9ddd 100644
--- a/.github/workflows/flink_cdc_base.yml
+++ b/.github/workflows/flink_cdc_base.yml
@@ -27,16 +27,16 @@ on:
description: "Flink versions to test against."
required: false
type: string
- default: "['generic']"
+ default: "['1.20.3']"
modules:
description: "Flink CDC modules to test against."
required: true
type: string
- parallelism:
- description: "Flink parallelism."
+ e2e-stages:
+ description: "Pipeline E2E stage filters (JSON array). Defaults to a
single empty entry meaning no filter (run everything). Set to e.g.
['stage1','stage2','stage3'] to fan out the pipeline E2E module across multiple
jobs by subpackage."
required: false
- type: number
- default: 4
+ type: string
+ default: '[""]'
custom-maven-parameter:
description: "Custom maven parameter."
required: false
@@ -44,6 +44,7 @@ on:
jobs:
test:
+ name: jdk${{ matrix.java-version }}-flink-${{ matrix.flink-version }}, ${{
matrix.module }}${{ matrix.stage != '' && format(', {0}', matrix.stage) || '' }}
runs-on: ubuntu-22.04
timeout-minutes: 120
strategy:
@@ -53,6 +54,7 @@ jobs:
java-version: ${{ fromJSON(inputs.java-versions) }}
flink-version: ${{ fromJSON(inputs.flink-versions) }}
module: ${{ fromJSON(inputs.modules) }}
+ stage: ${{ fromJSON(inputs.e2e-stages) }}
steps:
- run: echo "Running CI pipeline for JDK version ${{ matrix.java-version
}}"
- name: Clean up disk space
@@ -60,7 +62,7 @@ jobs:
set -euo pipefail
echo "Disk space before cleanup"
- df -h
+ df -h
echo "Cleaning up disk space"
sudo rm -rf /usr/share/dotnet
@@ -72,6 +74,25 @@ jobs:
echo "Disk space after cleanup"
df -h
+ - name: Relocate Docker data-root to /mnt
+ run: |
+ set -euo pipefail
+
+ # GitHub-hosted runners ship a large secondary volume mounted at /mnt
+ # (typically /dev/sdb1, ~75G). The default Docker data-root lives on
+ # /dev/root and gets crowded by Maven, build artifacts and other test
+ # containers, leaving e2e containers (notably OceanBase, which OBD
+ # pre-checks for ~10G free) failing to boot. Point Docker at /mnt to
+ # give containers the headroom they need.
+ sudo systemctl stop docker docker.socket
+ sudo mkdir -p /mnt/docker
+ echo '{"data-root":"/mnt/docker"}' | sudo tee /etc/docker/daemon.json
+ sudo systemctl start docker
+ docker info | grep "Docker Root Dir"
+
+ echo "Disk space after Docker relocation"
+ df -h
+
- name: Check out repository code
uses: actions/checkout@v6
with:
@@ -108,16 +129,27 @@ jobs:
modules=$(./.github/workflows/modules.py test "${{ matrix.module }}")
compile_modules=$(./.github/workflows/modules.py compile "${{
matrix.module }}")
-
+
build_maven_parameter="-DspecifiedMongoVersion=8.0.14"
if [ ! -z "${{ matrix.flink-version }}" ]; then
build_maven_parameter="${build_maven_parameter:+$build_maven_parameter
}-DspecifiedFlinkVersion=${{ matrix.flink-version }}"
fi
-
+
build_maven_parameter="${build_maven_parameter:+$build_maven_parameter }${{
inputs.custom-maven-parameter }}"
-
- mvn --no-snapshot-updates -B -DskipTests ${{
inputs.custom-maven-parameter }} -pl $compile_modules -am install && mvn
--no-snapshot-updates -B $build_maven_parameter -pl $modules
-DspecifiedParallelism=${{ inputs.parallelism }} -Duser.timezone=$jvm_timezone
verify
+
+ # Pipeline E2E stage filter: when matrix.stage is set, restrict
surefire to the
+ # matching subpackage. Stage1 hosts the YARN test; stage2/stage3
skip it via
+ # -DskipYarnTest=true. Empty stage (default) means no filter — run
everything.
+ e2e_stage="${{ matrix.stage }}"
+ if [ -n "$e2e_stage" ]; then
+ build_maven_parameter="$build_maven_parameter
-De2eIncludePattern=**/${e2e_stage}/**/*ITCase.java"
+ if [ "$e2e_stage" != "stage1" ]; then
+ build_maven_parameter="$build_maven_parameter
-DskipYarnTest=true"
+ fi
+ fi
+
+ mvn --no-snapshot-updates -B -DskipTests ${{
inputs.custom-maven-parameter }} -pl $compile_modules -am install && mvn
--no-snapshot-updates -B $build_maven_parameter -pl $modules
-Duser.timezone=$jvm_timezone verify
- name: Print JVM thread dumps when cancelled
if: ${{ failure() }}
diff --git a/.github/workflows/flink_cdc_ci.yml
b/.github/workflows/flink_cdc_ci.yml
index 1dd0a6b14c..97a682c089 100644
--- a/.github/workflows/flink_cdc_ci.yml
+++ b/.github/workflows/flink_cdc_ci.yml
@@ -100,32 +100,22 @@ jobs:
custom-maven-parameter: "-Pflink2"
modules: "['mysql-source', 'postgres-source, oceanbase-source, tidb,
vitess', 'oracle, sqlserver', 'db2, mongodb']"
pipeline_e2e:
- strategy:
- max-parallel: 2
- fail-fast: false
- matrix:
- parallelism: [ 1, 4 ]
- name: Pipeline E2E Tests (${{ matrix.parallelism }}-Parallelism)
+ name: Pipeline E2E Tests
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[11]"
flink-versions: "['1.20.3']"
modules: "['pipeline_e2e']"
- parallelism: ${{ matrix.parallelism }}
+ e2e-stages: "['stage1', 'stage2', 'stage3']"
pipeline_e2e_2_x:
- strategy:
- max-parallel: 2
- fail-fast: false
- matrix:
- parallelism: [ 1, 4 ]
- name: Pipeline E2E Tests 2.x (${{ matrix.parallelism }}-Parallelism)
+ name: Pipeline E2E Tests 2.x
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[11]"
flink-versions: "['2.2.0']"
custom-maven-parameter: "-Pflink2"
modules: "['pipeline_e2e']"
- parallelism: ${{ matrix.parallelism }}
+ e2e-stages: "['stage1', 'stage2', 'stage3']"
source_e2e:
name: Source E2E Tests
uses: ./.github/workflows/flink_cdc_base.yml
diff --git a/.github/workflows/flink_cdc_ci_nightly.yml
b/.github/workflows/flink_cdc_ci_nightly.yml
index fa1fa0f8ef..22f1ff5022 100644
--- a/.github/workflows/flink_cdc_ci_nightly.yml
+++ b/.github/workflows/flink_cdc_ci_nightly.yml
@@ -97,33 +97,23 @@ jobs:
modules: "['mysql-source', 'postgres-source, oceanbase-source, tidb,
vitess', 'oracle, sqlserver', 'db2, mongodb']"
pipeline_e2e:
if: github.repository == 'apache/flink-cdc'
- strategy:
- max-parallel: 2
- fail-fast: false
- matrix:
- parallelism: [ 1, 4 ]
- name: Pipeline E2E Tests (${{ matrix.parallelism }}-Parallelism)
+ name: Pipeline E2E Tests
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[17]"
flink-versions: "['1.20.3']"
modules: "['pipeline_e2e']"
- parallelism: ${{ matrix.parallelism }}
+ e2e-stages: "['stage1', 'stage2', 'stage3']"
pipeline_e2e_2_x:
if: github.repository == 'apache/flink-cdc'
- strategy:
- max-parallel: 2
- fail-fast: false
- matrix:
- parallelism: [ 1, 4 ]
- name: Pipeline E2E Tests 2.x (${{ matrix.parallelism }}-Parallelism)
+ name: Pipeline E2E Tests 2.x
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[17]"
flink-versions: "['2.2.0']"
custom-maven-parameter: "-Pflink2"
modules: "['pipeline_e2e']"
- parallelism: ${{ matrix.parallelism }}
+ e2e-stages: "['stage1', 'stage2', 'stage3']"
source_e2e:
if: github.repository == 'apache/flink-cdc'
name: Source E2E Tests
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
index 7e33a006d3..838656559e 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
@@ -43,6 +43,12 @@ limitations under the License.
<fluss.version>0.9.0-incubating</fluss.version>
<jmh.version>1.37</jmh.version>
<hudi.version>1.1.0</hudi.version>
+ <!-- E2E stage filtering: CI overrides via
-De2eIncludePattern=**/stageN/**/*ITCase.java
+ and -DskipYarnTest=true for stages where the YARN test should not
run. The defaults
+ here preserve the original `mvn verify` behavior for local
developers. -->
+ <e2eIncludePattern>**/*ITCase.java</e2eIncludePattern>
+
<e2eExcludePattern>**/MysqlE2eWithYarnApplicationITCase.java</e2eExcludePattern>
+ <skipYarnTest>false</skipYarnTest>
</properties>
<dependencies>
@@ -458,10 +464,10 @@ limitations under the License.
</goals>
<configuration>
<includes>
- <include>**/*.*</include>
+ <include>${e2eIncludePattern}</include>
</includes>
<excludes>
-
<exclude>**/MysqlE2eWithYarnApplicationITCase.java</exclude>
+ <exclude>${e2eExcludePattern}</exclude>
</excludes>
<forkCount>1</forkCount>
<systemPropertyVariables>
@@ -476,6 +482,7 @@ limitations under the License.
<goal>test</goal>
</goals>
<configuration>
+ <skip>${skipYarnTest}</skip>
<includes>
<include>**/MysqlE2eWithYarnApplicationITCase.java</include>
</includes>
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
index 7b551db833..b7583c53fe 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage1;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
@@ -24,6 +24,8 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,6 +38,8 @@ import java.time.Duration;
import java.util.function.Function;
/** End-to-end tests for mysql cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MysqlE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MysqlE2eITCase.class);
@@ -45,6 +49,10 @@ class MysqlE2eITCase extends PipelineTestEnvironment {
private final Function<String, String> dbNameFormatter =
(s) -> String.format(s, mysqlInventoryDatabase.getDatabaseName());
+ MysqlE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@BeforeEach
public void before() throws Exception {
super.before();
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eWithYarnApplicationITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eWithYarnApplicationITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eWithYarnApplicationITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eWithYarnApplicationITCase.java
index 7de50469b9..69973d1050 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eWithYarnApplicationITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eWithYarnApplicationITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage1;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
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/stage1/UdfE2eITCase.java
similarity index 97%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/UdfE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/UdfE2eITCase.java
index 24b76f7210..6ec9b6bf49 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/stage1/UdfE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage1;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.test.utils.TestUtils;
@@ -24,9 +24,11 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,12 +42,18 @@ import java.util.function.Function;
import java.util.stream.Stream;
/** E2e tests for User-defined functions. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class UdfE2eITCase extends PipelineTestEnvironment {
- private static final Logger LOG =
LoggerFactory.getLogger(TransformE2eITCase.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(UdfE2eITCase.class);
protected final UniqueDatabase udfTestDatabase =
new UniqueDatabase(MYSQL, "transform_test", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ UdfE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private final Function<String, String> dbNameFormatter =
(s) -> String.format(s, udfTestDatabase.getDatabaseName());
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/ValuesE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/ValuesE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/ValuesE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/ValuesE2eITCase.java
index 0f7bc28cbb..8cfc83e73e 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/ValuesE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/ValuesE2eITCase.java
@@ -15,20 +15,28 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage1;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
/** End-to-end tests for values cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class ValuesE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(ValuesE2eITCase.class);
+ ValuesE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@Test
void testValuesSingleSplitSingleTable() throws Exception {
String pipelineJob =
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/MySqlToPaimonMigrationITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/MySqlToPaimonMigrationITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/MySqlToPaimonMigrationITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/MySqlToPaimonMigrationITCase.java
index d94cdab266..0a697dff70 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/MySqlToPaimonMigrationITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/MySqlToPaimonMigrationITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.migration;
+package org.apache.flink.cdc.pipeline.tests.stage1.migration;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.test.utils.TestUtils;
@@ -29,8 +29,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -54,6 +56,8 @@ import java.util.stream.Collectors;
* E2e cases for stopping & restarting jobs of `MySQL source to Paimon sink`
from previous state.
*/
@EnabledIfSystemProperty(named = "specifiedFlinkVersion", matches = "^1.*")
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MySqlToPaimonMigrationITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlToPaimonMigrationITCase.class);
@@ -61,6 +65,11 @@ class MySqlToPaimonMigrationITCase extends
PipelineTestEnvironment {
private static final Duration PAIMON_TESTCASE_TIMEOUT =
Duration.ofMinutes(3);
protected UniqueDatabase mysqlInventoryDatabase;
+
+ MySqlToPaimonMigrationITCase(int parallelism) {
+ super(parallelism);
+ }
+
private final Function<String, String> dbNameFormatter =
(s) -> String.format(s, mysqlInventoryDatabase.getDatabaseName());
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/YamlJobMigrationITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/YamlJobMigrationITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/YamlJobMigrationITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/YamlJobMigrationITCase.java
index ef21129351..898ac7976a 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/YamlJobMigrationITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/migration/YamlJobMigrationITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.migration;
+package org.apache.flink.cdc.pipeline.tests.stage1.migration;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
@@ -30,8 +30,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,6 +47,8 @@ import java.util.function.Function;
/** E2e cases for stopping & restarting jobs from previous state. */
@EnabledIfSystemProperty(named = "specifiedFlinkVersion", matches = "^1.*")
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class YamlJobMigrationITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(YamlJobMigrationITCase.class);
@@ -53,6 +57,10 @@ class YamlJobMigrationITCase extends PipelineTestEnvironment
{
private final Function<String, String> dbNameFormatter =
(s) -> String.format(s, mysqlInventoryDatabase.getDatabaseName());
+ YamlJobMigrationITCase(int parallelism) {
+ super(parallelism);
+ }
+
@BeforeEach
public void before() throws Exception {
super.before();
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/CheckStep.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/CheckStep.java
similarity index 95%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/CheckStep.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/CheckStep.java
index f7e6acb96c..ec01aca8b4 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/CheckStep.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/CheckStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
import java.util.List;
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/ExecStep.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/ExecStep.java
similarity index 94%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/ExecStep.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/ExecStep.java
index 45bca3988b..7e6dd9003d 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/ExecStep.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/ExecStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
/** A {@link SpecStep} to execute SQL commands. */
class ExecStep implements SpecStep {
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/FlinkPipelineSpecsITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/FlinkPipelineSpecsITCase.java
similarity index 96%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/FlinkPipelineSpecsITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/FlinkPipelineSpecsITCase.java
index b7ac0507ce..4c209cb5eb 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/FlinkPipelineSpecsITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/FlinkPipelineSpecsITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
import org.apache.flink.cdc.cli.utils.YamlParserUtils;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
@@ -23,9 +23,11 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,11 +52,17 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
/** YAML-spec based pipeline test cases. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class FlinkPipelineSpecsITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkPipelineSpecsITCase.class);
private static final String MYSQL_CONTAINER_ALIAS = "mysql";
+ public FlinkPipelineSpecsITCase(int parallelism) {
+ super(parallelism);
+ }
+
static Stream<Arguments> loadTestSpecs() throws IOException {
Path path =
Paths.get("").toAbsolutePath().resolve("src/test/resources/rules");
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/RuleSpec.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/RuleSpec.java
similarity index 95%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/RuleSpec.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/RuleSpec.java
index 087749545e..02a4049bbf 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/RuleSpec.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/RuleSpec.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
import java.util.List;
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SpecStep.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SpecStep.java
similarity index 93%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SpecStep.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SpecStep.java
index 8e53e8849f..879f7b8018 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SpecStep.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SpecStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
/** A single step of {@link RuleSpec}. */
public interface SpecStep {}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SubmitStep.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SubmitStep.java
similarity index 96%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SubmitStep.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SubmitStep.java
index 435a98f162..46fd8dbbad 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/SubmitStep.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/specs/SubmitStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.specs;
+package org.apache.flink.cdc.pipeline.tests.stage1.specs;
import java.util.List;
import java.util.stream.Collectors;
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/AiFunctionE2eITCase.java
similarity index 95%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/AiFunctionE2eITCase.java
index 8264898974..e03ac0dde2 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/AiFunctionE2eITCase.java
@@ -15,19 +15,27 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import java.nio.file.Path;
import java.time.Duration;
/** E2e tests for AI functions with the dummy model SPI. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class AiFunctionE2eITCase extends PipelineTestEnvironment {
+ AiFunctionE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static final String TABLE_1 =
"default_namespace.default_schema.table1";
private static final String TABLE_2 =
"default_namespace.default_schema.table2";
private static final String DUMMY_TEXT_RESULTS =
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/FlussE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/FlussE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/FlussE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/FlussE2eITCase.java
index 8f12e20c15..81c14d5589 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/FlussE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/FlussE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
@@ -26,6 +26,7 @@ import
org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
@@ -49,6 +50,8 @@ import java.util.stream.Collectors;
/** An End-to-end test case for Fluss pipeline connector. */
@Testcontainers
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class FlussE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(FlussE2eITCase.class);
@@ -56,6 +59,10 @@ public class FlussE2eITCase extends PipelineTestEnvironment {
private static final String flussImageTag =
"apache/fluss:0.9.0-incubating";
private static final String zooKeeperImageTag = "zookeeper:3.9.2";
+ public FlussE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static final List<String> flussCoordinatorProperties =
Arrays.asList(
"zookeeper.address: zookeeper:2181",
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToDorisE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToDorisE2eITCase.java
index b9b8aaccbe..51e429a22c 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToDorisE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.doris.sink.utils.DorisContainer;
@@ -27,6 +27,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
@@ -49,9 +51,15 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
/** End-to-end tests for mysql cdc to Doris pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MySqlToDorisE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlToDorisE2eITCase.class);
+ MySqlToDorisE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@Container
protected static final DorisContainer DORIS =
new DorisContainer(NETWORK).withNetworkAliases("doris");
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToHudiE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToHudiE2eITCase.java
index 45ae95d2aa..ec1e96291b 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToHudiE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
@@ -38,6 +38,8 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.BindMode;
@@ -69,12 +71,18 @@ import java.util.stream.Stream;
/** End-to-end tests for mysql cdc to Iceberg pipeline job. */
@EnabledIfSystemProperty(named = "specifiedFlinkVersion", matches = "^1.*")
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class MySqlToHudiE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlToHudiE2eITCase.class);
private static final Duration HUDI_TESTCASE_TIMEOUT =
Duration.ofMinutes(20);
+ public MySqlToHudiE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static final String FLINK_LIB_DIR = "/opt/flink/lib";
private static final String PEEK_SQL_FILE = "peek-hudi.sql";
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToIcebergE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToIcebergE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToIcebergE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToIcebergE2eITCase.java
index 2239353f42..526e1a774e 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToIcebergE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToIcebergE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.test.utils.TestUtils;
@@ -38,6 +38,8 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.BindMode;
@@ -68,12 +70,18 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
/** End-to-end tests for mysql cdc to Iceberg pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class MySqlToIcebergE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlToIcebergE2eITCase.class);
@TempDir public Path temporaryFolder;
+ public MySqlToIcebergE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
protected final UniqueDatabase inventoryDatabase =
new UniqueDatabase(MYSQL, "iceberg_inventory", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToPaimonE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToPaimonE2eITCase.java
index 7ba68669bb..0b12122360 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MySqlToPaimonE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
@@ -26,6 +26,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -48,11 +50,17 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
/** End-to-end tests for mysql cdc to Paimon pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MySqlToPaimonE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlToPaimonE2eITCase.class);
private static final Duration PAIMON_TESTCASE_TIMEOUT =
Duration.ofMinutes(3);
+ MySqlToPaimonE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
protected final UniqueDatabase inventoryDatabase =
new UniqueDatabase(MYSQL, "paimon_inventory", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
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/stage2/MysqlToKafkaE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlToKafkaE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToKafkaE2eITCase.java
index f02ba986b9..9bb82ebabe 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/stage2/MysqlToKafkaE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.event.TableId;
@@ -39,6 +39,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.KafkaContainer;
@@ -72,9 +74,15 @@ import static
org.apache.flink.util.DockerImageVersions.KAFKA;
import static org.assertj.core.api.Assertions.assertThat;
/** End-to-end tests for mysql cdc to Kafka pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MysqlToKafkaE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MysqlToKafkaE2eITCase.class);
+ MysqlToKafkaE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static AdminClient admin;
private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
private static final int ZK_TIMEOUT_MILLIS = 30000;
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
similarity index 97%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
index 66c40f8205..873d518cc7 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
@@ -23,6 +23,8 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -39,10 +41,16 @@ import java.util.function.Function;
import java.util.stream.Collectors;
/** E2e tests for pipelines that use Python UDFs. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class PythonUdfE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(PythonUdfE2eITCase.class);
+ PythonUdfE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static final String CONTAINER_PYTHON_EXECUTABLE =
"/usr/bin/python3";
private static final String PYTHON_FILES_DIRECTORY =
"/opt/flink/python-deps";
private static final String PEMJA_VERSION = "0.5.7";
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/stage2/SqlServerE2eITCase.java
similarity index 97%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SqlServerE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/SqlServerE2eITCase.java
index ad24ff2755..5e31354dc3 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/stage2/SqlServerE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Deadline;
@@ -26,6 +26,8 @@ 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.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container.ExecResult;
@@ -53,9 +55,15 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
/** End-to-end tests for sqlserver cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class SqlServerE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(SqlServerE2eITCase.class);
+ public SqlServerE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private static final Pattern COMMENT_PATTERN =
Pattern.compile("^(.*)--.*$");
private static final String INTER_CONTAINER_SQL_SERVER_ALIAS =
"mssqlserver";
private static final String MSSQL_SERVER_IMAGE =
"mcr.microsoft.com/mssql/server:2019-latest";
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MaxComputeE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
similarity index 96%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MaxComputeE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
index 59d966bc2a..3d31ba015a 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MaxComputeE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/MaxComputeE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.time.Deadline;
@@ -34,6 +34,8 @@ import com.aliyun.odps.task.SQLTask;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
@@ -56,9 +58,15 @@ import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
/** End-to-end tests for maxcompute cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class MaxComputeE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(MaxComputeE2eITCase.class);
+ MaxComputeE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
public static final DockerImageName MAXCOMPUTE_IMAGE =
DockerImageName.parse("maxcompute/maxcompute-emulator:v0.0.7");
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OceanBaseE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OceanBaseE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OceanBaseE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OceanBaseE2eITCase.java
index 2bebedb3bb..e5b22d26d3 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OceanBaseE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OceanBaseE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
@@ -29,6 +29,8 @@ import org.apache.commons.text.StringEscapeUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -51,11 +53,17 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
/** OceanBase flink cdc pipeline connector sink integrate test. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class OceanBaseE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(OceanBaseE2eITCase.class);
private static final String MYSQL_TEST_TABLE_NAME = "products";
+ OceanBaseE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@Container
private static final OceanBaseContainer OB_SERVER =
OceanBaseTestUtils.createOceanBaseContainerForJdbc()
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OracleE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OracleE2eITCase.java
index 498012e54d..8567a1b224 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/OracleE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.oracle.source.OracleSourceTestBase;
@@ -27,6 +27,8 @@ 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.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.DockerClientFactory;
@@ -61,10 +63,17 @@ import static
org.apache.flink.cdc.connectors.oracle.source.OracleSourceTestBase
import static
org.apache.flink.cdc.connectors.oracle.source.OracleSourceTestBase.TOP_USER;
/** End-to-end tests for Oracle cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class OracleE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(OracleE2eITCase.class);
protected static final Pattern COMMENT_PATTERN =
Pattern.compile("^(.*)--.*$");
private static OracleContainer oracle;
+
+ public OracleE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
//
------------------------------------------------------------------------------------------
// Oracle Variables (we always use Oracle as the data source for easier
verifying)
//
------------------------------------------------------------------------------------------
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PostgresE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/PostgresE2eITCase.java
similarity index 97%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PostgresE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/PostgresE2eITCase.java
index d6d368c94e..0dec676db8 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PostgresE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/PostgresE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
@@ -24,6 +24,8 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.PostgreSQLContainer;
@@ -42,9 +44,15 @@ import static
org.apache.flink.cdc.connectors.postgres.PostgresTestBase.getJdbcC
import static
org.apache.flink.cdc.connectors.postgres.PostgresTestBase.getSlotName;
/** End-to-end tests for postgres cdc pipeline job. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
public class PostgresE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(PostgresE2eITCase.class);
+ public PostgresE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
//
------------------------------------------------------------------------------------------
// Postgres Variables (we always use Postgres as the data source for
easier verifying)
//
------------------------------------------------------------------------------------------
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/RouteE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/RouteE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java
index 49ef42c4e2..bb5a2c97fb 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/RouteE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
@@ -23,6 +23,7 @@ import
org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
@@ -37,9 +38,15 @@ import java.util.function.Function;
import java.util.stream.IntStream;
/** E2e tests for routing features. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class RouteE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(RouteE2eITCase.class);
+ RouteE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
protected static final int TEST_TABLE_NUMBER = 60;
protected final UniqueDatabase routeTestDatabase =
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolveE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolveE2eITCase.java
index 68068e38a6..03e2963e2f 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolveE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
@@ -24,6 +24,8 @@ 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.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,12 +38,18 @@ import java.util.Collections;
import java.util.List;
/** E2e tests for Schema Evolution cases. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class SchemaEvolveE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(SchemaEvolveE2eITCase.class);
protected final UniqueDatabase schemaEvolveDatabase =
new UniqueDatabase(MYSQL, "schema_evolve", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ SchemaEvolveE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@BeforeEach
public void before() throws Exception {
super.before();
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolvingTransformE2eITCase.java
similarity index 98%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolvingTransformE2eITCase.java
index 1ba2be8f13..f17afc2818 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/SchemaEvolvingTransformE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
@@ -24,6 +24,8 @@ 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.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,6 +38,8 @@ import java.util.Collections;
import java.util.List;
/** E2e tests for Schema Evolution cases. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class SchemaEvolvingTransformE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(SchemaEvolvingTransformE2eITCase.class);
@@ -43,6 +47,10 @@ class SchemaEvolvingTransformE2eITCase extends
PipelineTestEnvironment {
protected final UniqueDatabase schemaEvolveDatabase =
new UniqueDatabase(MYSQL, "schema_evolve", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ SchemaEvolvingTransformE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
@BeforeEach
public void before() throws Exception {
super.before();
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/stage3/TransformE2eITCase.java
similarity index 99%
rename from
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
rename to
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/TransformE2eITCase.java
index 4f9c7bd19a..a844dae14e 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/stage3/TransformE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests;
+package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.data.DateData;
@@ -28,6 +28,7 @@ import
org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
@@ -52,12 +53,18 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
/** E2e tests for the {@link PreTransformOperator} and {@link
PostTransformOperator}. */
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
class TransformE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG =
LoggerFactory.getLogger(TransformE2eITCase.class);
protected final UniqueDatabase transformTestDatabase =
new UniqueDatabase(MYSQL, "transform_test", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ TransformE2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
private final Function<String, String> dbNameFormatter =
(s) -> {
String databaseName = transformTestDatabase.getDatabaseName();
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 2ee0b97c75..f4db8e3202 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
@@ -85,17 +85,10 @@ public abstract class PipelineTestEnvironment extends
TestLogger {
private static final Logger LOG =
LoggerFactory.getLogger(PipelineTestEnvironment.class);
- protected Integer parallelism = getParallelism();
+ protected final Integer parallelism;
- private int getParallelism() {
- try {
- return
Integer.parseInt(System.getProperty("specifiedParallelism"));
- } catch (NumberFormatException ex) {
- LOG.warn(
- "Unable to parse specified parallelism configuration ({}
provided). Use 4 by default.",
- System.getProperty("specifiedParallelism"));
- return 4;
- }
+ protected PipelineTestEnvironment(int parallelism) {
+ this.parallelism = parallelism;
}
//
------------------------------------------------------------------------------------------
