This is an automated email from the ASF dual-hosted git repository.
zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 0be9a4889af Refactor PipelineJobOption (#37119)
0be9a4889af is described below
commit 0be9a4889afc57c8c65c184a69b369e89085ceec
Author: Liang Zhang <[email protected]>
AuthorDate: Sat Nov 15 22:51:53 2025 +0800
Refactor PipelineJobOption (#37119)
* Refactor PipelineJobOption
* Refactor PipelineJobOption
---
.../pipeline/core/job/type/PipelineJobOption.java | 13 ++++++--
.../core/job/type/PipelineJobOptionTest.java | 38 ++++++++++++++++++++++
.../data/pipeline/cdc/CDCJobType.java | 3 +-
.../consistencycheck/ConsistencyCheckJobType.java | 4 +--
.../scenario/migration/MigrationJobType.java | 4 +--
5 files changed, 52 insertions(+), 10 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOption.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOption.java
index dde24fc09bc..05032330787 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOption.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOption.java
@@ -21,7 +21,9 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.data.pipeline.core.job.PipelineJob;
import
org.apache.shardingsphere.data.pipeline.core.job.config.yaml.swapper.YamlPipelineJobConfigurationSwapper;
+import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlConsistencyCheckJobItemProgressSwapper;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlPipelineJobItemProgressSwapper;
+import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlTransmissionJobItemProgressSwapper;
/**
* Pipeline job type.
@@ -36,8 +38,6 @@ public final class PipelineJobOption {
private final YamlPipelineJobConfigurationSwapper<?, ?>
yamlJobConfigurationSwapper;
- private final YamlPipelineJobItemProgressSwapper<?, ?>
yamlJobItemProgressSwapper;
-
private final Class<? extends PipelineJob> jobClass;
private final boolean
isIgnoreToStartDisabledJobWhenJobItemProgressIsFinished;
@@ -47,4 +47,13 @@ public final class PipelineJobOption {
private final String toBeStoppedPreviousJobType;
private final boolean isForceNoShardingWhenConvertToJobConfigurationPOJO;
+
+ /**
+ * Get YAML job item progress swapper.
+ *
+ * @return YAML job item progress swapper
+ */
+ public YamlPipelineJobItemProgressSwapper<?, ?>
getYamlJobItemProgressSwapper() {
+ return isTransmissionJob ? new
YamlTransmissionJobItemProgressSwapper() : new
YamlConsistencyCheckJobItemProgressSwapper();
+ }
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOptionTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOptionTest.java
new file mode 100644
index 00000000000..270beac6c0a
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/type/PipelineJobOptionTest.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.job.type;
+
+import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlConsistencyCheckJobItemProgressSwapper;
+import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlTransmissionJobItemProgressSwapper;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.isA;
+
+class PipelineJobOptionTest {
+
+ @Test
+ void assertGetYamlJobItemProgressSwapperWithTransmissionJob() {
+ assertThat(new PipelineJobOption("00", true, null, null, false, null,
null, false).getYamlJobItemProgressSwapper(),
isA(YamlTransmissionJobItemProgressSwapper.class));
+ }
+
+ @Test
+ void assertGetYamlJobItemProgressSwapperWithNotTransmissionJob() {
+ assertThat(new PipelineJobOption("00", false, null, null, false, null,
null, false).getYamlJobItemProgressSwapper(),
isA(YamlConsistencyCheckJobItemProgressSwapper.class));
+ }
+}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobType.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobType.java
index 2ce1ea04835..7f0555c9707 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobType.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobType.java
@@ -19,7 +19,6 @@ package org.apache.shardingsphere.data.pipeline.cdc;
import org.apache.shardingsphere.data.pipeline.cdc.config.CDCJobConfiguration;
import
org.apache.shardingsphere.data.pipeline.cdc.config.yaml.swapper.YamlCDCJobConfigurationSwapper;
-import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlTransmissionJobItemProgressSwapper;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobOption;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
import org.apache.shardingsphere.data.pipeline.core.pojo.PipelineJobObjective;
@@ -31,7 +30,7 @@ public final class CDCJobType implements
PipelineJobType<CDCJobConfiguration> {
@Override
public PipelineJobOption getOption() {
- return new PipelineJobOption("03", true, new
YamlCDCJobConfigurationSwapper(), new YamlTransmissionJobItemProgressSwapper(),
CDCJob.class, false, null, null, true);
+ return new PipelineJobOption("03", true, new
YamlCDCJobConfigurationSwapper(), CDCJob.class, false, null, null, true);
}
@Override
diff --git
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobType.java
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobType.java
index f33c104727b..2811e463546 100644
---
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobType.java
+++
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobType.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.scenario.consistencycheck;
import
org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfiguration;
-import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlConsistencyCheckJobItemProgressSwapper;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobOption;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
import org.apache.shardingsphere.data.pipeline.core.pojo.PipelineJobObjective;
@@ -31,8 +30,7 @@ public final class ConsistencyCheckJobType implements
PipelineJobType<PipelineJo
@Override
public PipelineJobOption getOption() {
- return new PipelineJobOption("02",
- false, new YamlConsistencyCheckJobConfigurationSwapper(), new
YamlConsistencyCheckJobItemProgressSwapper(), ConsistencyCheckJob.class, true,
null, null, false);
+ return new PipelineJobOption("02", false, new
YamlConsistencyCheckJobConfigurationSwapper(), ConsistencyCheckJob.class, true,
null, null, false);
}
@Override
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobType.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobType.java
index 6119d3cde43..c409b95f2b7 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobType.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobType.java
@@ -20,7 +20,6 @@ package
org.apache.shardingsphere.data.pipeline.scenario.migration;
import
org.apache.shardingsphere.data.pipeline.core.consistencycheck.ConsistencyCheckJobItemProgressContext;
import
org.apache.shardingsphere.data.pipeline.core.consistencycheck.PipelineDataConsistencyChecker;
import
org.apache.shardingsphere.data.pipeline.core.context.TransmissionProcessContext;
-import
org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.swapper.YamlTransmissionJobItemProgressSwapper;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobOption;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
import org.apache.shardingsphere.data.pipeline.core.pojo.PipelineJobObjective;
@@ -38,8 +37,7 @@ public final class MigrationJobType implements
PipelineJobType<MigrationJobConfi
@Override
public PipelineJobOption getOption() {
- return new PipelineJobOption("01",
- true, new YamlMigrationJobConfigurationSwapper(), new
YamlTransmissionJobItemProgressSwapper(), MigrationJob.class, false,
"CONSISTENCY_CHECK", "CONSISTENCY_CHECK", false);
+ return new PipelineJobOption("01", true, new
YamlMigrationJobConfigurationSwapper(), MigrationJob.class, false,
"CONSISTENCY_CHECK", "CONSISTENCY_CHECK", false);
}
@Override
