You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2023/11/18 09:11:41 UTC
(shardingsphere) branch master updated: Add YamlPipelineJobItemProgressConfiguration (#29074)
This is an automated email from the ASF dual-hosted git repository.
panjuan 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 19c6551e3c5 Add YamlPipelineJobItemProgressConfiguration (#29074)
19c6551e3c5 is described below
commit 19c6551e3c540dcf49b554859fe9627007303f72
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sat Nov 18 17:11:35 2023 +0800
Add YamlPipelineJobItemProgressConfiguration (#29074)
---
.../yaml/YamlConsistencyCheckJobItemProgress.java | 4 ++--
.../YamlConsistencyCheckJobItemProgressSwapper.java | 5 +++++
.../YamlInventoryIncrementalJobItemProgress.java | 4 ++--
...mlInventoryIncrementalJobItemProgressSwapper.java | 5 +++++
.../core/job/service/InventoryIncrementalJobAPI.java | 5 +----
.../pipeline/core/job/service/PipelineJobAPI.java | 14 +++-----------
.../core/job/service/PipelineJobManager.java | 19 +++++++++++++++++--
.../impl/AbstractInventoryIncrementalJobAPIImpl.java | 13 ++++---------
...=> YamlPipelineJobItemProgressConfiguration.java} | 9 ++-------
.../job/yaml/YamlPipelineJobItemProgressSwapper.java | 12 +++++++++---
.../data/pipeline/cdc/api/impl/CDCJobAPI.java | 3 ++-
.../data/pipeline/cdc/core/job/CDCJob.java | 2 +-
.../pipeline/cdc/core/prepare/CDCJobPreparer.java | 2 +-
.../consistencycheck/ConsistencyCheckJob.java | 4 +++-
.../api/impl/ConsistencyCheckJobAPI.java | 20 +++++++++-----------
.../pipeline/scenario/migration/MigrationJob.java | 5 ++++-
.../migration/prepare/MigrationJobPreparer.java | 2 +-
.../migration/api/impl/MigrationJobAPITest.java | 2 +-
18 files changed, 72 insertions(+), 58 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
index 2ed595408b3..72ea55e1865 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
@@ -19,7 +19,7 @@ package org.apache.shardingsphere.data.pipeline.common.job.progress.yaml;
import lombok.Getter;
import lombok.Setter;
-import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressConfiguration;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -29,7 +29,7 @@ import java.util.Map;
*/
@Getter
@Setter
-public final class YamlConsistencyCheckJobItemProgress implements YamlConfiguration {
+public final class YamlConsistencyCheckJobItemProgress implements YamlPipelineJobItemProgressConfiguration {
private String status;
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
index 066df10fc28..eb582d1d896 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
@@ -50,4 +50,9 @@ public final class YamlConsistencyCheckJobItemProgressSwapper implements YamlPip
result.setStatus(JobStatus.valueOf(yamlConfig.getStatus()));
return result;
}
+
+ @Override
+ public Class<YamlConsistencyCheckJobItemProgress> getYamlProgressClass() {
+ return YamlConsistencyCheckJobItemProgress.class;
+ }
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgress.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgress.java
index e73596ce667..1e44dcf8984 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgress.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgress.java
@@ -19,14 +19,14 @@ package org.apache.shardingsphere.data.pipeline.common.job.progress.yaml;
import lombok.Getter;
import lombok.Setter;
-import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressConfiguration;
/**
* YAML inventory incremental job item progress.
*/
@Getter
@Setter
-public final class YamlInventoryIncrementalJobItemProgress implements YamlConfiguration {
+public final class YamlInventoryIncrementalJobItemProgress implements YamlPipelineJobItemProgressConfiguration {
private String status;
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgressSwapper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgressSwapper.java
index 24c2f6a5a74..1b29d4f52de 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgressSwapper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlInventoryIncrementalJobItemProgressSwapper.java
@@ -57,4 +57,9 @@ public final class YamlInventoryIncrementalJobItemProgressSwapper implements Yam
result.setInventoryRecordsCount(yamlProgress.getInventoryRecordsCount());
return result;
}
+
+ @Override
+ public Class<YamlInventoryIncrementalJobItemProgress> getYamlProgressClass() {
+ return YamlInventoryIncrementalJobItemProgress.class;
+ }
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/InventoryIncrementalJobAPI.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/InventoryIncrementalJobAPI.java
index 908c3103c76..fa4594f7eab 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/InventoryIncrementalJobAPI.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/InventoryIncrementalJobAPI.java
@@ -37,13 +37,13 @@ import java.sql.SQLException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
/**
* Inventory incremental job API.
*/
public interface InventoryIncrementalJobAPI extends PipelineJobAPI {
+ @SuppressWarnings("unchecked")
@Override
default YamlInventoryIncrementalJobItemProgressSwapper getYamlJobItemProgressSwapper() {
return new YamlInventoryIncrementalJobItemProgressSwapper();
@@ -123,9 +123,6 @@ public interface InventoryIncrementalJobAPI extends PipelineJobAPI {
*/
Map<Integer, InventoryIncrementalJobItemProgress> getJobProgress(PipelineJobConfiguration pipelineJobConfig);
- @Override
- Optional<InventoryIncrementalJobItemProgress> getJobItemProgress(String jobId, int shardingItem);
-
/**
* Get job infos.
*
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java
index b5ab26690a0..168b18d88ad 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java
@@ -21,6 +21,7 @@ import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.common.job.PipelineJob;
import org.apache.shardingsphere.data.pipeline.common.job.progress.PipelineJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobConfigurationSwapper;
+import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressSwapper;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
@@ -43,10 +44,10 @@ public interface PipelineJobAPI extends TypedSPI {
/**
* Get YAML pipeline job item progress swapper.
*
+ * @param <T> type of pipeline job item progress
* @return YAML pipeline job item progress swapper
*/
- @SuppressWarnings("rawtypes")
- YamlPipelineJobItemProgressSwapper getYamlJobItemProgressSwapper();
+ <T extends PipelineJobItemProgress> YamlPipelineJobItemProgressSwapper<YamlPipelineJobItemProgressConfiguration, T> getYamlJobItemProgressSwapper();
/**
* Whether to ignore to start disabled job when job item progress is finished.
@@ -75,15 +76,6 @@ public interface PipelineJobAPI extends TypedSPI {
return Optional.empty();
}
- /**
- * Get job item progress.
- *
- * @param jobId job id
- * @param shardingItem sharding item
- * @return job item progress, may be null
- */
- Optional<? extends PipelineJobItemProgress> getJobItemProgress(String jobId, int shardingItem);
-
/**
* Update job item status.
*
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
index 99d846dd6d7..e7fd3ad85e6 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -32,6 +32,8 @@ import org.apache.shardingsphere.data.pipeline.common.util.PipelineDistributedBa
import org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobCreationWithInvalidShardingCountException;
import org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobHasAlreadyStartedException;
import org.apache.shardingsphere.data.pipeline.core.job.PipelineJobIdUtils;
+import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.job.yaml.YamlPipelineJobItemProgressSwapper;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
@@ -95,7 +97,7 @@ public final class PipelineJobManager {
*/
public void startDisabledJob(final String jobId) {
if (jobAPI.isIgnoreToStartDisabledJobWhenJobItemProgressIsFinished()) {
- Optional<? extends PipelineJobItemProgress> jobItemProgress = jobAPI.getJobItemProgress(jobId, 0);
+ Optional<? extends PipelineJobItemProgress> jobItemProgress = getJobItemProgress(jobId, 0);
if (jobItemProgress.isPresent() && JobStatus.FINISHED == jobItemProgress.get().getStatus()) {
log.info("job status is FINISHED, ignore, jobId={}", jobId);
return;
@@ -201,6 +203,20 @@ public final class PipelineJobManager {
.filter(each -> jobType.equals(PipelineJobIdUtils.parseJobType(each.getJobName()).getType()));
}
+ /**
+ * Get job item progress.
+ *
+ * @param jobId job id
+ * @param shardingItem sharding item
+ * @param <T> type of pipeline job item progress
+ * @return job item progress, may be null
+ */
+ public <T extends PipelineJobItemProgress> Optional<T> getJobItemProgress(final String jobId, final int shardingItem) {
+ YamlPipelineJobItemProgressSwapper<YamlPipelineJobItemProgressConfiguration, T> swapper = jobAPI.getYamlJobItemProgressSwapper();
+ Optional<String> progress = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProgress(jobId, shardingItem);
+ return progress.map(optional -> swapper.swapToObject(YamlEngine.unmarshal(optional, swapper.getYamlProgressClass(), true)));
+ }
+
/**
* Persist job item progress.
*
@@ -221,7 +237,6 @@ public final class PipelineJobManager {
.updateJobItemProgress(jobItemContext.getJobId(), jobItemContext.getShardingItem(), convertJobItemProgress(jobItemContext));
}
- @SuppressWarnings("unchecked")
private String convertJobItemProgress(final PipelineJobItemContext jobItemContext) {
return YamlEngine.marshal(jobAPI.getYamlJobItemProgressSwapper().swapToYamlConfiguration(jobItemContext.toProgress()));
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractInventoryIncrementalJobAPIImpl.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractInventoryIncrementalJobAPIImpl.java
index 1706b2c4c9f..6b0cf4171e6 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractInventoryIncrementalJobAPIImpl.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractInventoryIncrementalJobAPIImpl.java
@@ -25,7 +25,6 @@ import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextKey
import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.common.job.progress.InventoryIncrementalJobItemProgress;
import org.apache.shardingsphere.data.pipeline.common.job.progress.JobOffsetInfo;
-import org.apache.shardingsphere.data.pipeline.common.job.progress.yaml.YamlInventoryIncrementalJobItemProgress;
import org.apache.shardingsphere.data.pipeline.common.job.progress.yaml.YamlJobOffsetInfo;
import org.apache.shardingsphere.data.pipeline.common.job.progress.yaml.YamlJobOffsetInfoSwapper;
import org.apache.shardingsphere.data.pipeline.common.pojo.DataConsistencyCheckAlgorithmInfo;
@@ -77,10 +76,11 @@ public abstract class AbstractInventoryIncrementalJobAPIImpl implements Inventor
@Override
public Map<Integer, InventoryIncrementalJobItemProgress> getJobProgress(final PipelineJobConfiguration jobConfig) {
+ PipelineJobManager jobManager = new PipelineJobManager(this);
String jobId = jobConfig.getJobId();
JobConfigurationPOJO jobConfigPOJO = PipelineJobIdUtils.getElasticJobConfigurationPOJO(jobId);
return IntStream.range(0, jobConfig.getJobShardingCount()).boxed().collect(LinkedHashMap::new, (map, each) -> {
- Optional<InventoryIncrementalJobItemProgress> jobItemProgress = getJobItemProgress(jobId, each);
+ Optional<InventoryIncrementalJobItemProgress> jobItemProgress = jobManager.getJobItemProgress(jobId, each);
jobItemProgress.ifPresent(optional -> optional.setActive(!jobConfigPOJO.isDisabled()));
map.put(each, jobItemProgress.orElse(null));
}, LinkedHashMap::putAll);
@@ -130,15 +130,10 @@ public abstract class AbstractInventoryIncrementalJobAPIImpl implements Inventor
return jobOffsetInfoSwapper.swapToObject(new YamlJobOffsetInfo());
}
- @Override
- public Optional<InventoryIncrementalJobItemProgress> getJobItemProgress(final String jobId, final int shardingItem) {
- Optional<String> progress = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProgress(jobId, shardingItem);
- return progress.map(optional -> getYamlJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(optional, YamlInventoryIncrementalJobItemProgress.class)));
- }
-
@Override
public void updateJobItemStatus(final String jobId, final int shardingItem, final JobStatus status) {
- Optional<InventoryIncrementalJobItemProgress> jobItemProgress = getJobItemProgress(jobId, shardingItem);
+ PipelineJobManager jobManager = new PipelineJobManager(this);
+ Optional<InventoryIncrementalJobItemProgress> jobItemProgress = jobManager.getJobItemProgress(jobId, shardingItem);
if (!jobItemProgress.isPresent()) {
return;
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressConfiguration.java
similarity index 66%
copy from kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java
copy to kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressConfiguration.java
index 90f522b0910..1a1ef05cbdb 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressConfiguration.java
@@ -17,15 +17,10 @@
package org.apache.shardingsphere.data.pipeline.core.job.yaml;
-import org.apache.shardingsphere.data.pipeline.common.job.progress.PipelineJobItemProgress;
import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
-import org.apache.shardingsphere.infra.util.yaml.swapper.YamlConfigurationSwapper;
/**
- * YAML pipeline job configuration swapper.
- *
- * @param <Y> type of YAML configuration
- * @param <T> type of swapped pipeline job item progress
+ * YAML pipeline job item progress configuration.
*/
-public interface YamlPipelineJobItemProgressSwapper<Y extends YamlConfiguration, T extends PipelineJobItemProgress> extends YamlConfigurationSwapper<Y, T> {
+public interface YamlPipelineJobItemProgressConfiguration extends YamlConfiguration {
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java
index 90f522b0910..4f803634df9 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/yaml/YamlPipelineJobItemProgressSwapper.java
@@ -18,14 +18,20 @@
package org.apache.shardingsphere.data.pipeline.core.job.yaml;
import org.apache.shardingsphere.data.pipeline.common.job.progress.PipelineJobItemProgress;
-import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
import org.apache.shardingsphere.infra.util.yaml.swapper.YamlConfigurationSwapper;
/**
* YAML pipeline job configuration swapper.
*
- * @param <Y> type of YAML configuration
+ * @param <Y> type of YAML pipeline job item progress configuration
* @param <T> type of swapped pipeline job item progress
*/
-public interface YamlPipelineJobItemProgressSwapper<Y extends YamlConfiguration, T extends PipelineJobItemProgress> extends YamlConfigurationSwapper<Y, T> {
+public interface YamlPipelineJobItemProgressSwapper<Y extends YamlPipelineJobItemProgressConfiguration, T extends PipelineJobItemProgress> extends YamlConfigurationSwapper<Y, T> {
+
+ /**
+ * Get YAML pipeline job item progress configuration class.
+ *
+ * @return YAML pipeline job item progress configuration class
+ */
+ Class<Y> getYamlProgressClass();
}
diff --git a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
index 070328d868e..e4aec40e5cb 100644
--- a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
+++ b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
@@ -168,9 +168,10 @@ public final class CDCJobAPI extends AbstractInventoryIncrementalJobAPIImpl {
private void initIncrementalPosition(final CDCJobConfiguration jobConfig) {
String jobId = jobConfig.getJobId();
+ PipelineJobManager jobManager = new PipelineJobManager(this);
try (PipelineDataSourceManager pipelineDataSourceManager = new DefaultPipelineDataSourceManager()) {
for (int i = 0; i < jobConfig.getJobShardingCount(); i++) {
- if (getJobItemProgress(jobId, i).isPresent()) {
+ if (jobManager.getJobItemProgress(jobId, i).isPresent()) {
continue;
}
IncrementalDumperContext dumperContext = buildDumperContext(jobConfig, i, new TableAndSchemaNameMapper(jobConfig.getSchemaTableNames()));
diff --git a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/job/CDCJob.java b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/job/CDCJob.java
index c2637c0de36..657b7b48ccd 100644
--- a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/job/CDCJob.java
+++ b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/job/CDCJob.java
@@ -106,7 +106,7 @@ public final class CDCJob extends AbstractPipelineJob implements SimpleJob {
}
private CDCJobItemContext buildPipelineJobItemContext(final CDCJobConfiguration jobConfig, final int shardingItem) {
- Optional<InventoryIncrementalJobItemProgress> initProgress = jobAPI.getJobItemProgress(jobConfig.getJobId(), shardingItem);
+ Optional<InventoryIncrementalJobItemProgress> initProgress = jobManager.getJobItemProgress(jobConfig.getJobId(), shardingItem);
CDCProcessContext jobProcessContext = jobAPI.buildPipelineProcessContext(jobConfig);
CDCTaskConfiguration taskConfig = jobAPI.buildTaskConfiguration(jobConfig, shardingItem, jobProcessContext.getPipelineProcessConfig());
return new CDCJobItemContext(jobConfig, shardingItem, initProgress.orElse(null), jobProcessContext, taskConfig, dataSourceManager, sink);
diff --git a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java
index e79887f33cd..31f50757bbe 100644
--- a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java
+++ b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java
@@ -89,7 +89,7 @@ public final class CDCJobPreparer {
private void initTasks0(final CDCJobItemContext jobItemContext, final AtomicBoolean inventoryImporterUsed, final List<CDCChannelProgressPair> inventoryChannelProgressPairs,
final AtomicBoolean incrementalImporterUsed, final List<CDCChannelProgressPair> incrementalChannelProgressPairs) {
- Optional<InventoryIncrementalJobItemProgress> jobItemProgress = jobAPI.getJobItemProgress(jobItemContext.getJobId(), jobItemContext.getShardingItem());
+ Optional<InventoryIncrementalJobItemProgress> jobItemProgress = jobManager.getJobItemProgress(jobItemContext.getJobId(), jobItemContext.getShardingItem());
if (!jobItemProgress.isPresent()) {
jobManager.persistJobItemProgress(jobItemContext);
}
diff --git a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJob.java b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJob.java
index 7d289dd0a89..913fb82ed29 100644
--- a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJob.java
+++ b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJob.java
@@ -22,6 +22,7 @@ import org.apache.shardingsphere.data.pipeline.common.context.PipelineJobItemCon
import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.common.job.progress.ConsistencyCheckJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.AbstractSimplePipelineJob;
+import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
import org.apache.shardingsphere.data.pipeline.core.task.runner.PipelineTasksRunner;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.api.impl.ConsistencyCheckJobAPI;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.config.ConsistencyCheckJobConfiguration;
@@ -46,7 +47,8 @@ public final class ConsistencyCheckJob extends AbstractSimplePipelineJob {
public ConsistencyCheckJobItemContext buildPipelineJobItemContext(final ShardingContext shardingContext) {
ConsistencyCheckJobConfiguration jobConfig = new YamlConsistencyCheckJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
ConsistencyCheckJobAPI jobAPI = (ConsistencyCheckJobAPI) getJobAPI();
- Optional<ConsistencyCheckJobItemProgress> jobItemProgress = jobAPI.getJobItemProgress(jobConfig.getJobId(), shardingContext.getShardingItem());
+ PipelineJobManager jobManager = new PipelineJobManager(jobAPI);
+ Optional<ConsistencyCheckJobItemProgress> jobItemProgress = jobManager.getJobItemProgress(jobConfig.getJobId(), shardingContext.getShardingItem());
return new ConsistencyCheckJobItemContext(jobConfig, shardingContext.getShardingItem(), JobStatus.RUNNING, jobItemProgress.orElse(null));
}
diff --git a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
index 62a1520c06b..4ba4d05079d 100644
--- a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
+++ b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
@@ -22,7 +22,6 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextKey;
import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.common.job.progress.ConsistencyCheckJobItemProgress;
-import org.apache.shardingsphere.data.pipeline.common.job.progress.yaml.YamlConsistencyCheckJobItemProgress;
import org.apache.shardingsphere.data.pipeline.common.job.progress.yaml.YamlConsistencyCheckJobItemProgressSwapper;
import org.apache.shardingsphere.data.pipeline.common.pojo.ConsistencyCheckJobItemInfo;
import org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.GovernanceRepositoryAPI;
@@ -84,7 +83,8 @@ public final class ConsistencyCheckJobAPI implements PipelineJobAPI {
GovernanceRepositoryAPI repositoryAPI = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(parentJobId));
Optional<String> latestCheckJobId = repositoryAPI.getLatestCheckJobId(parentJobId);
if (latestCheckJobId.isPresent()) {
- Optional<ConsistencyCheckJobItemProgress> progress = getJobItemProgress(latestCheckJobId.get(), 0);
+ PipelineJobManager jobManager = new PipelineJobManager(this);
+ Optional<ConsistencyCheckJobItemProgress> progress = jobManager.getJobItemProgress(latestCheckJobId.get(), 0);
if (!progress.isPresent() || JobStatus.FINISHED != progress.get().getStatus()) {
log.info("check job already exists and status is not FINISHED, progress={}", progress);
throw new UncompletedConsistencyCheckJobExistsException(latestCheckJobId.get());
@@ -117,15 +117,10 @@ public final class ConsistencyCheckJobAPI implements PipelineJobAPI {
return true;
}
- @Override
- public Optional<ConsistencyCheckJobItemProgress> getJobItemProgress(final String jobId, final int shardingItem) {
- Optional<String> progress = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProgress(jobId, shardingItem);
- return progress.map(s -> getYamlJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(s, YamlConsistencyCheckJobItemProgress.class, true)));
- }
-
@Override
public void updateJobItemStatus(final String jobId, final int shardingItem, final JobStatus status) {
- Optional<ConsistencyCheckJobItemProgress> jobItemProgress = getJobItemProgress(jobId, shardingItem);
+ PipelineJobManager jobManager = new PipelineJobManager(this);
+ Optional<ConsistencyCheckJobItemProgress> jobItemProgress = jobManager.getJobItemProgress(jobId, shardingItem);
if (!jobItemProgress.isPresent()) {
log.warn("updateJobItemStatus, jobProgress is null, jobId={}, shardingItem={}", jobId, shardingItem);
return;
@@ -193,7 +188,8 @@ public final class ConsistencyCheckJobAPI implements PipelineJobAPI {
Optional<String> latestCheckJobId = governanceRepositoryAPI.getLatestCheckJobId(parentJobId);
ShardingSpherePreconditions.checkState(latestCheckJobId.isPresent(), () -> new ConsistencyCheckJobNotFoundException(parentJobId));
String checkJobId = latestCheckJobId.get();
- Optional<ConsistencyCheckJobItemProgress> progress = getJobItemProgress(checkJobId, 0);
+ PipelineJobManager jobManager = new PipelineJobManager(this);
+ Optional<ConsistencyCheckJobItemProgress> progress = jobManager.getJobItemProgress(checkJobId, 0);
if (!progress.isPresent()) {
return Collections.emptyList();
}
@@ -233,7 +229,8 @@ public final class ConsistencyCheckJobAPI implements PipelineJobAPI {
Optional<String> latestCheckJobId = governanceRepositoryAPI.getLatestCheckJobId(parentJobId);
ShardingSpherePreconditions.checkState(latestCheckJobId.isPresent(), () -> new ConsistencyCheckJobNotFoundException(parentJobId));
String checkJobId = latestCheckJobId.get();
- Optional<ConsistencyCheckJobItemProgress> progress = getJobItemProgress(checkJobId, 0);
+ PipelineJobManager jobManager = new PipelineJobManager(this);
+ Optional<ConsistencyCheckJobItemProgress> progress = jobManager.getJobItemProgress(checkJobId, 0);
ConsistencyCheckJobItemInfo result = new ConsistencyCheckJobItemInfo();
JobConfigurationPOJO jobConfigPOJO = PipelineJobIdUtils.getElasticJobConfigurationPOJO(checkJobId);
result.setActive(!jobConfigPOJO.isDisabled());
@@ -306,6 +303,7 @@ public final class ConsistencyCheckJobAPI implements PipelineJobAPI {
return new YamlConsistencyCheckJobConfigurationSwapper();
}
+ @SuppressWarnings("unchecked")
@Override
public YamlConsistencyCheckJobItemProgressSwapper getYamlJobItemProgressSwapper() {
return new YamlConsistencyCheckJobItemProgressSwapper();
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java
index adef5960815..136842f0541 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java
@@ -24,6 +24,7 @@ import org.apache.shardingsphere.data.pipeline.common.datasource.DefaultPipeline
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.common.job.progress.InventoryIncrementalJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.AbstractSimplePipelineJob;
+import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
import org.apache.shardingsphere.data.pipeline.core.task.runner.InventoryIncrementalTasksRunner;
import org.apache.shardingsphere.data.pipeline.core.task.runner.PipelineTasksRunner;
import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.MigrationJobAPI;
@@ -46,6 +47,8 @@ public final class MigrationJob extends AbstractSimplePipelineJob {
private final MigrationJobAPI jobAPI = new MigrationJobAPI();
+ private final PipelineJobManager jobManager = new PipelineJobManager(jobAPI);
+
private final PipelineDataSourceManager dataSourceManager = new DefaultPipelineDataSourceManager();
// Shared by all sharding items
@@ -59,7 +62,7 @@ public final class MigrationJob extends AbstractSimplePipelineJob {
protected InventoryIncrementalJobItemContext buildPipelineJobItemContext(final ShardingContext shardingContext) {
int shardingItem = shardingContext.getShardingItem();
MigrationJobConfiguration jobConfig = new YamlMigrationJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
- Optional<InventoryIncrementalJobItemProgress> initProgress = jobAPI.getJobItemProgress(shardingContext.getJobName(), shardingItem);
+ Optional<InventoryIncrementalJobItemProgress> initProgress = jobManager.getJobItemProgress(shardingContext.getJobName(), shardingItem);
MigrationProcessContext jobProcessContext = jobAPI.buildPipelineProcessContext(jobConfig);
MigrationTaskConfiguration taskConfig = jobAPI.buildTaskConfiguration(jobConfig, shardingItem, jobProcessContext.getPipelineProcessConfig());
return new MigrationJobItemContext(jobConfig, shardingItem, initProgress.orElse(null), jobProcessContext, taskConfig, dataSourceManager);
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
index 37772491969..d473001b873 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
@@ -125,7 +125,7 @@ public final class MigrationJobPreparer {
MigrationJobConfiguration jobConfig = jobItemContext.getJobConfig();
String jobId = jobConfig.getJobId();
LockContext lockContext = PipelineContextManager.getContext(PipelineJobIdUtils.parseContextKey(jobId)).getContextManager().getInstanceContext().getLockContext();
- if (!jobAPI.getJobItemProgress(jobId, jobItemContext.getShardingItem()).isPresent()) {
+ if (!jobManager.getJobItemProgress(jobId, jobItemContext.getShardingItem()).isPresent()) {
jobManager.persistJobItemProgress(jobItemContext);
}
LockDefinition lockDefinition = new GlobalLockDefinition(String.format(GlobalLockNames.PREPARE.getLockName(), jobConfig.getJobId()));
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/migration/api/impl/MigrationJobAPITest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/migration/api/impl/MigrationJobAPITest.java
index fd0e33cdfa0..a1a1c905dbd 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/migration/api/impl/MigrationJobAPITest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/migration/api/impl/MigrationJobAPITest.java
@@ -247,7 +247,7 @@ class MigrationJobAPITest {
MigrationJobItemContext jobItemContext = PipelineContextUtils.mockMigrationJobItemContext(jobConfig);
jobManager.persistJobItemProgress(jobItemContext);
jobAPI.updateJobItemStatus(jobConfig.getJobId(), 0, JobStatus.FINISHED);
- Optional<InventoryIncrementalJobItemProgress> actual = jobAPI.getJobItemProgress(jobItemContext.getJobId(), jobItemContext.getShardingItem());
+ Optional<InventoryIncrementalJobItemProgress> actual = jobManager.getJobItemProgress(jobItemContext.getJobId(), jobItemContext.getShardingItem());
assertTrue(actual.isPresent());
assertThat(actual.get().getStatus(), is(JobStatus.FINISHED));
}