This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng 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 4b834438e60 Extract list() method to PipelineJobPublicAPI and
AbstractPipelineJobAPIImpl for common usage (#20309)
4b834438e60 is described below
commit 4b834438e606d29e2012318c75edefcb0fbd699e
Author: Da Xiang Huang <[email protected]>
AuthorDate: Sat Aug 20 11:20:42 2022 +0800
Extract list() method to PipelineJobPublicAPI and
AbstractPipelineJobAPIImpl for common usage (#20309)
---
.../data/pipeline/api/MigrationJobPublicAPI.java | 4 +--
.../data/pipeline/api/PipelineJobPublicAPI.java | 10 ++++++
.../MigrationJobInfo.java} | 31 ++++------------
.../PipelineJobInfo.java} | 42 +++++++++++-----------
.../core/api/impl/AbstractPipelineJobAPIImpl.java | 26 ++++++++++++++
.../data/pipeline/core/job/FinishedCheckJob.java | 6 ++--
.../scenario/migration/MigrationJobAPIImpl.java | 21 +++--------
.../scenario/rulealtered/RuleAlteredJobWorker.java | 4 +--
.../core/fixture/MigrationJobAPIFixture.java | 4 +--
.../core/api/impl/MigrationJobAPIImplTest.java | 10 +++---
.../pipeline/core/job/FinishedCheckJobTest.java | 6 ++--
11 files changed, 83 insertions(+), 81 deletions(-)
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/MigrationJobPublicAPI.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/MigrationJobPublicAPI.java
index ce1e9f18b28..f370af21eba 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/MigrationJobPublicAPI.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/MigrationJobPublicAPI.java
@@ -20,7 +20,7 @@ package org.apache.shardingsphere.data.pipeline.api;
import
org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsistencyCheckResult;
import
org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
import
org.apache.shardingsphere.data.pipeline.api.pojo.DataConsistencyCheckAlgorithmInfo;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import org.apache.shardingsphere.infra.datasource.props.DataSourceProperties;
import org.apache.shardingsphere.infra.util.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.util.spi.type.required.RequiredSPI;
@@ -41,7 +41,7 @@ public interface MigrationJobPublicAPI extends
PipelineJobPublicAPI, RequiredSPI
*
* @return job infos
*/
- List<JobInfo> list();
+ List<PipelineJobInfo> list();
/**
* Get job progress.
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
index dd8c9175c2f..13c9aa5de7e 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
@@ -17,8 +17,11 @@
package org.apache.shardingsphere.data.pipeline.api;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import org.apache.shardingsphere.infra.util.spi.type.typed.TypedSPI;
+import java.util.List;
+
/**
* Pipeline job public API.
*/
@@ -44,4 +47,11 @@ public interface PipelineJobPublicAPI extends TypedSPI {
* @param jobId job id
*/
void remove(String jobId);
+
+ /**
+ * Get pipeline job info.
+ *
+ * @return jobInfos
+ */
+ List<PipelineJobInfo> list();
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/MigrationJobInfo.java
similarity index 59%
copy from
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
copy to
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/MigrationJobInfo.java
index dd8c9175c2f..a781f28be1d 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/MigrationJobInfo.java
@@ -15,33 +15,14 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.api;
-
-import org.apache.shardingsphere.infra.util.spi.type.typed.TypedSPI;
+package org.apache.shardingsphere.data.pipeline.api.pojo;
/**
- * Pipeline job public API.
+ * Migration job info.
*/
-public interface PipelineJobPublicAPI extends TypedSPI {
-
- /**
- * Start disabled job.
- *
- * @param jobId job id
- */
- void startDisabledJob(String jobId);
-
- /**
- * Stop pipeline job.
- *
- * @param jobId job id
- */
- void stop(String jobId);
+public class MigrationJobInfo extends PipelineJobInfo {
- /**
- * Remove pipeline job.
- *
- * @param jobId job id
- */
- void remove(String jobId);
+ public MigrationJobInfo(final String jobId) {
+ super(jobId);
+ }
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/PipelineJobInfo.java
similarity index 60%
copy from
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
copy to
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/PipelineJobInfo.java
index dd8c9175c2f..ea939cd08d4 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineJobPublicAPI.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/PipelineJobInfo.java
@@ -15,33 +15,31 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.api;
+package org.apache.shardingsphere.data.pipeline.api.pojo;
-import org.apache.shardingsphere.infra.util.spi.type.typed.TypedSPI;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
/**
- * Pipeline job public API.
+ * Pipeline Job Info.
*/
-public interface PipelineJobPublicAPI extends TypedSPI {
+@RequiredArgsConstructor
+@Getter
+@Setter
+public abstract class PipelineJobInfo {
- /**
- * Start disabled job.
- *
- * @param jobId job id
- */
- void startDisabledJob(String jobId);
+ private final String jobId;
- /**
- * Stop pipeline job.
- *
- * @param jobId job id
- */
- void stop(String jobId);
+ private boolean active;
- /**
- * Remove pipeline job.
- *
- * @param jobId job id
- */
- void remove(String jobId);
+ private String tables;
+
+ private int shardingTotalCount;
+
+ private String createTime;
+
+ private String stopTime;
+
+ private transient String jobParameter;
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/AbstractPipelineJobAPIImpl.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/AbstractPipelineJobAPIImpl.java
index e07fb3035d6..f2f6a3cb583 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/AbstractPipelineJobAPIImpl.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/AbstractPipelineJobAPIImpl.java
@@ -17,13 +17,19 @@
package org.apache.shardingsphere.data.pipeline.core.api.impl;
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import lombok.extern.slf4j.Slf4j;
import
org.apache.shardingsphere.data.pipeline.api.config.job.PipelineJobConfiguration;
import
org.apache.shardingsphere.data.pipeline.api.config.job.yaml.YamlPipelineJobConfiguration;
import org.apache.shardingsphere.data.pipeline.api.job.PipelineJobId;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineJobAPI;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobNotFoundException;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineVerifyFailedException;
@@ -32,6 +38,8 @@ import
org.apache.shardingsphere.data.pipeline.core.metadata.node.PipelineMetaDa
import
org.apache.shardingsphere.data.pipeline.core.util.PipelineDistributedBarrier;
import org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJob;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
+import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
+import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import java.time.LocalDateTime;
@@ -56,6 +64,24 @@ public abstract class AbstractPipelineJobAPIImpl implements
PipelineJobAPI {
protected abstract String marshalJobIdLeftPart(PipelineJobId
pipelineJobId);
+ @Override
+ public List<PipelineJobInfo> list() {
+ checkModeConfig();
+ return getJobBriefInfos().map(each ->
getJobInfo(each.getJobName())).collect(Collectors.toList());
+ }
+
+ protected void checkModeConfig() {
+ ModeConfiguration modeConfig = PipelineContext.getModeConfig();
+ Preconditions.checkNotNull(modeConfig, "Mode configuration is
required.");
+
Preconditions.checkArgument("Cluster".equalsIgnoreCase(modeConfig.getType()),
"Mode must be `Cluster`.");
+ }
+
+ private Stream<JobBriefInfo> getJobBriefInfos() {
+ return
PipelineAPIFactory.getJobStatisticsAPI().getAllJobsBriefInfo().stream().filter(each
-> !each.getJobName().startsWith("_"));
+ }
+
+ protected abstract PipelineJobInfo getJobInfo(String jobName);
+
@Override
public Optional<String> start(final PipelineJobConfiguration jobConfig) {
String jobId = jobConfig.getJobId();
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
index c491cfa11ef..76ecceabbf5 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
@@ -23,7 +23,7 @@ import
org.apache.shardingsphere.data.pipeline.api.config.job.yaml.YamlMigration
import
org.apache.shardingsphere.data.pipeline.api.detect.RuleAlteredJobAlmostCompletedParameter;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import
org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobAPI;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobAPIFactory;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationProcessContext;
@@ -45,8 +45,8 @@ public final class FinishedCheckJob implements SimpleJob {
// TODO only one proxy node could do data consistency check in proxy
cluster
@Override
public void execute(final ShardingContext shardingContext) {
- List<JobInfo> jobInfos = jobAPI.list();
- for (JobInfo jobInfo : jobInfos) {
+ List<PipelineJobInfo> jobInfos = jobAPI.list();
+ for (PipelineJobInfo jobInfo : jobInfos) {
if (!jobInfo.isActive()) {
continue;
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
index ba19be8209e..06ce962775c 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
@@ -17,7 +17,6 @@
package org.apache.shardingsphere.data.pipeline.scenario.migration;
-import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Hex;
@@ -36,7 +35,8 @@ import
org.apache.shardingsphere.data.pipeline.api.job.JobType;
import org.apache.shardingsphere.data.pipeline.api.job.PipelineJobId;
import
org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
import
org.apache.shardingsphere.data.pipeline.api.pojo.DataConsistencyCheckAlgorithmInfo;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.MigrationJobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineJobItemAPI;
@@ -57,7 +57,6 @@ import
org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorit
import
org.apache.shardingsphere.data.pipeline.spi.rulealtered.RuleAlteredJobConfigurationPreparerFactory;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
-import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
import
org.apache.shardingsphere.infra.config.rule.data.pipeline.PipelineProcessConfiguration;
import
org.apache.shardingsphere.infra.config.rule.data.pipeline.PipelineReadConfiguration;
import org.apache.shardingsphere.infra.datasource.props.DataSourceProperties;
@@ -159,24 +158,12 @@ public final class MigrationJobAPIImpl extends
AbstractPipelineJobAPIImpl implem
return new MigrationProcessContext(pipelineJobConfig.getJobId(),
processConfig);
}
- @Override
- public List<JobInfo> list() {
- checkModeConfig();
- return getJobBriefInfos().map(each ->
getJobInfo(each.getJobName())).collect(Collectors.toList());
- }
-
- private void checkModeConfig() {
- ModeConfiguration modeConfig = PipelineContext.getModeConfig();
- Preconditions.checkNotNull(modeConfig, "Mode configuration is
required.");
-
Preconditions.checkArgument("Cluster".equalsIgnoreCase(modeConfig.getType()),
"Mode must be `Cluster`.");
- }
-
private Stream<JobBriefInfo> getJobBriefInfos() {
return
PipelineAPIFactory.getJobStatisticsAPI().getAllJobsBriefInfo().stream().filter(each
-> !each.getJobName().startsWith("_"));
}
- private JobInfo getJobInfo(final String jobName) {
- JobInfo result = new JobInfo(jobName);
+ protected PipelineJobInfo getJobInfo(final String jobName) {
+ MigrationJobInfo result = new MigrationJobInfo(jobName);
JobConfigurationPOJO jobConfigPOJO =
getElasticJobConfigPOJO(result.getJobId());
MigrationJobConfiguration jobConfig =
getJobConfiguration(jobConfigPOJO);
result.setActive(!jobConfigPOJO.isDisabled());
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
index 1ffee2397fa..7d81e4c0960 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
@@ -30,7 +30,7 @@ import
org.apache.shardingsphere.data.pipeline.api.datasource.config.impl.Shardi
import
org.apache.shardingsphere.data.pipeline.api.datasource.config.yaml.YamlPipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.JobType;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory;
import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException;
@@ -230,7 +230,7 @@ public final class RuleAlteredJobWorker {
private boolean hasUncompletedJobOfSameDatabaseName(final String
databaseName) {
boolean result = false;
- for (JobInfo each : MigrationJobAPIFactory.getInstance().list()) {
+ for (PipelineJobInfo each :
MigrationJobAPIFactory.getInstance().list()) {
if
(MigrationJobAPIFactory.getInstance().getJobProgress(each.getJobId()).values().stream()
.allMatch(progress -> null != progress &&
progress.getStatus().equals(JobStatus.FINISHED))) {
continue;
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/MigrationJobAPIFixture.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/MigrationJobAPIFixture.java
index 5b313dbac2a..81e59f59ed7 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/MigrationJobAPIFixture.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/MigrationJobAPIFixture.java
@@ -27,7 +27,7 @@ import
org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.PipelineJobId;
import
org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
import
org.apache.shardingsphere.data.pipeline.api.pojo.DataConsistencyCheckAlgorithmInfo;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobAPI;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationProcessContext;
import
org.apache.shardingsphere.infra.config.rule.data.pipeline.PipelineProcessConfiguration;
@@ -63,7 +63,7 @@ public final class MigrationJobAPIFixture implements
MigrationJobAPI {
}
@Override
- public List<JobInfo> list() {
+ public List<PipelineJobInfo> list() {
return null;
}
diff --git
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/api/impl/MigrationJobAPIImplTest.java
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/api/impl/MigrationJobAPIImplTest.java
index 16dad9490b5..f56f0ce16bc 100644
---
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/api/impl/MigrationJobAPIImplTest.java
+++
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/api/impl/MigrationJobAPIImplTest.java
@@ -28,7 +28,7 @@ import
org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDat
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.JobType;
import
org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineResourceAPI;
@@ -79,18 +79,18 @@ public final class MigrationJobAPIImplTest {
public void assertStartAndList() {
Optional<String> jobId =
jobAPI.start(JobConfigurationBuilder.createJobConfiguration());
assertTrue(jobId.isPresent());
- JobInfo jobInfo = getNonNullJobInfo(jobId.get());
+ PipelineJobInfo jobInfo = getNonNullJobInfo(jobId.get());
assertTrue(jobInfo.isActive());
assertThat(jobInfo.getTables(), is("t_order"));
assertThat(jobInfo.getShardingTotalCount(), is(1));
}
- private Optional<JobInfo> getJobInfo(final String jobId) {
+ private Optional<PipelineJobInfo> getJobInfo(final String jobId) {
return jobAPI.list().stream().filter(each ->
Objects.equals(each.getJobId(), jobId)).reduce((a, b) -> a);
}
- private JobInfo getNonNullJobInfo(final String jobId) {
- Optional<JobInfo> result = getJobInfo(jobId);
+ private PipelineJobInfo getNonNullJobInfo(final String jobId) {
+ Optional<PipelineJobInfo> result = getJobInfo(jobId);
assertTrue(result.isPresent());
return result.get();
}
diff --git
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJobTest.java
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJobTest.java
index 913ff7f65d7..400cc87014c 100644
---
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJobTest.java
+++
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJobTest.java
@@ -18,7 +18,7 @@
package org.apache.shardingsphere.data.pipeline.core.job;
import lombok.SneakyThrows;
-import org.apache.shardingsphere.data.pipeline.api.pojo.JobInfo;
+import org.apache.shardingsphere.data.pipeline.api.pojo.PipelineJobInfo;
import
org.apache.shardingsphere.data.pipeline.core.util.JobConfigurationBuilder;
import org.apache.shardingsphere.data.pipeline.core.util.PipelineContextUtil;
import org.apache.shardingsphere.data.pipeline.core.util.ReflectionUtil;
@@ -67,7 +67,7 @@ public final class FinishedCheckJobTest {
public void assertExecuteAllDisabledJob() {
Optional<String> jobId =
MigrationJobAPIFactory.getInstance().start(JobConfigurationBuilder.createJobConfiguration());
assertTrue(jobId.isPresent());
- List<JobInfo> jobInfos = MigrationJobAPIFactory.getInstance().list();
+ List<PipelineJobInfo> jobInfos =
MigrationJobAPIFactory.getInstance().list();
jobInfos.forEach(each -> each.setActive(false));
when(jobAPI.list()).thenReturn(jobInfos);
finishedCheckJob.execute(null);
@@ -77,7 +77,7 @@ public final class FinishedCheckJobTest {
public void assertExecuteActiveJob() {
Optional<String> jobId =
MigrationJobAPIFactory.getInstance().start(JobConfigurationBuilder.createJobConfiguration());
assertTrue(jobId.isPresent());
- List<JobInfo> jobInfos = MigrationJobAPIFactory.getInstance().list();
+ List<PipelineJobInfo> jobInfos =
MigrationJobAPIFactory.getInstance().list();
jobInfos.forEach(each -> each.setActive(true));
when(jobAPI.list()).thenReturn(jobInfos);
finishedCheckJob.execute(null);
