You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by ji...@apache.org on 2023/12/11 15:13:26 UTC
(shardingsphere) branch master updated: Add more generic types of AbstractSeparablePipelineJob (#29366)
This is an automated email from the ASF dual-hosted git repository.
jianglongtao 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 37fb5a8aab0 Add more generic types of AbstractSeparablePipelineJob (#29366)
37fb5a8aab0 is described below
commit 37fb5a8aab098dc66c73ad5b8f4fe74e9dfe7a94
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Mon Dec 11 23:13:19 2023 +0800
Add more generic types of AbstractSeparablePipelineJob (#29366)
---
.../core/job/AbstractSeparablePipelineJob.java | 23 +++++++++++-----------
.../consistencycheck/ConsistencyCheckJob.java | 8 +++-----
.../pipeline/scenario/migration/MigrationJob.java | 11 ++++-------
3 files changed, 19 insertions(+), 23 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java
index d5cbcd1c1de..ab6ab115ee8 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java
@@ -28,7 +28,6 @@ import org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfig
import org.apache.shardingsphere.data.pipeline.core.job.engine.PipelineJobRunnerManager;
import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
import org.apache.shardingsphere.data.pipeline.core.job.progress.PipelineJobItemProgress;
-import org.apache.shardingsphere.data.pipeline.core.job.progress.TransmissionJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobConfigurationManager;
import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobItemManager;
import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
@@ -41,12 +40,14 @@ import java.sql.SQLException;
/**
* Abstract separable pipeline job.
*
- * @param <T> type of pipeline job item context
+ * @param <T> type of pipeline job configuration
+ * @param <I> type of pipeline job item context
+ * @param <P> type of pipeline job item progress
*/
@RequiredArgsConstructor
@Getter
@Slf4j
-public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemContext> implements PipelineJob {
+public abstract class AbstractSeparablePipelineJob<T extends PipelineJobConfiguration, I extends PipelineJobItemContext, P extends PipelineJobItemProgress> implements PipelineJob {
private final PipelineJobRunnerManager jobRunnerManager;
@@ -65,9 +66,9 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
}
PipelineJobType jobType = PipelineJobIdUtils.parseJobType(jobId);
PipelineJobConfigurationManager jobConfigManager = new PipelineJobConfigurationManager(jobType);
- PipelineJobConfiguration jobConfig = jobConfigManager.getJobConfiguration(jobId);
- PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager = new PipelineJobItemManager<>(jobType.getYamlJobItemProgressSwapper());
- TransmissionJobItemProgress jobItemProgress = jobItemManager.getProgress(shardingContext.getJobName(), shardingItem).orElse(null);
+ T jobConfig = jobConfigManager.getJobConfiguration(jobId);
+ PipelineJobItemManager<P> jobItemManager = new PipelineJobItemManager<>(jobType.getYamlJobItemProgressSwapper());
+ P jobItemProgress = jobItemManager.getProgress(shardingContext.getJobName(), shardingItem).orElse(null);
try {
execute(buildJobItemContext(jobConfig, shardingItem, jobItemProgress));
// CHECKSTYLE:OFF
@@ -78,7 +79,7 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
}
}
- private void execute(final T jobItemContext) {
+ private void execute(final I jobItemContext) {
int shardingItem = jobItemContext.getShardingItem();
PipelineTasksRunner tasksRunner = buildTasksRunner(jobItemContext);
if (!jobRunnerManager.addTasksRunner(shardingItem, tasksRunner)) {
@@ -91,11 +92,11 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
tasksRunner.start();
}
- protected abstract T buildJobItemContext(PipelineJobConfiguration jobConfig, int shardingItem, PipelineJobItemProgress jobItemProgress);
+ protected abstract I buildJobItemContext(T jobConfig, int shardingItem, P jobItemProgress);
- protected abstract PipelineTasksRunner buildTasksRunner(T jobItemContext);
+ protected abstract PipelineTasksRunner buildTasksRunner(I jobItemContext);
- protected final void prepare(final T jobItemContext) {
+ protected final void prepare(final I jobItemContext) {
try {
doPrepare(jobItemContext);
// CHECKSTYLE:OFF
@@ -105,7 +106,7 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
}
}
- protected abstract void doPrepare(T jobItemContext) throws SQLException;
+ protected abstract void doPrepare(I jobItemContext) throws SQLException;
private void processFailed(final String jobId, final int shardingItem, final Exception ex) {
log.error("Job execution failed, {}-{}", jobId, shardingItem, ex);
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 cea091dd3fc..85b9c6f91cc 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
@@ -19,9 +19,7 @@ package org.apache.shardingsphere.data.pipeline.scenario.consistencycheck;
import org.apache.shardingsphere.data.pipeline.core.job.AbstractSeparablePipelineJob;
import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
-import org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.ConsistencyCheckJobItemProgress;
-import org.apache.shardingsphere.data.pipeline.core.job.progress.PipelineJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.task.runner.PipelineTasksRunner;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.config.ConsistencyCheckJobConfiguration;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.context.ConsistencyCheckJobItemContext;
@@ -30,11 +28,11 @@ import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.task.Co
/**
* Consistency check job.
*/
-public final class ConsistencyCheckJob extends AbstractSeparablePipelineJob<ConsistencyCheckJobItemContext> {
+public final class ConsistencyCheckJob extends AbstractSeparablePipelineJob<ConsistencyCheckJobConfiguration, ConsistencyCheckJobItemContext, ConsistencyCheckJobItemProgress> {
@Override
- public ConsistencyCheckJobItemContext buildJobItemContext(final PipelineJobConfiguration jobConfig, final int shardingItem, final PipelineJobItemProgress jobItemProgress) {
- return new ConsistencyCheckJobItemContext((ConsistencyCheckJobConfiguration) jobConfig, shardingItem, JobStatus.RUNNING, (ConsistencyCheckJobItemProgress) jobItemProgress);
+ public ConsistencyCheckJobItemContext buildJobItemContext(final ConsistencyCheckJobConfiguration jobConfig, final int shardingItem, final ConsistencyCheckJobItemProgress jobItemProgress) {
+ return new ConsistencyCheckJobItemContext(jobConfig, shardingItem, JobStatus.RUNNING, jobItemProgress);
}
@Override
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 feece1a82a4..42ea6849ded 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
@@ -26,9 +26,7 @@ import org.apache.shardingsphere.data.pipeline.core.importer.ImporterConfigurati
import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.IncrementalDumperContext;
import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.mapper.TableAndSchemaNameMapper;
import org.apache.shardingsphere.data.pipeline.core.job.AbstractSeparablePipelineJob;
-import org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
-import org.apache.shardingsphere.data.pipeline.core.job.progress.PipelineJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.progress.TransmissionJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.PipelineProcessConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.PipelineProcessConfigurationUtils;
@@ -59,20 +57,19 @@ import java.util.stream.Collectors;
* Migration job.
*/
@Slf4j
-public final class MigrationJob extends AbstractSeparablePipelineJob<MigrationJobItemContext> {
+public final class MigrationJob extends AbstractSeparablePipelineJob<MigrationJobConfiguration, MigrationJobItemContext, TransmissionJobItemProgress> {
private final PipelineProcessConfigurationPersistService processConfigPersistService = new PipelineProcessConfigurationPersistService();
private final MigrationJobPreparer jobPreparer = new MigrationJobPreparer();
@Override
- protected MigrationJobItemContext buildJobItemContext(final PipelineJobConfiguration jobConfig, final int shardingItem, final PipelineJobItemProgress jobItemProgress) {
+ protected MigrationJobItemContext buildJobItemContext(final MigrationJobConfiguration jobConfig, final int shardingItem, final TransmissionJobItemProgress jobItemProgress) {
PipelineProcessConfiguration processConfig = PipelineProcessConfigurationUtils.convertWithDefaultValue(
processConfigPersistService.load(PipelineJobIdUtils.parseContextKey(jobConfig.getJobId()), "MIGRATION"));
TransmissionProcessContext jobProcessContext = new TransmissionProcessContext(jobConfig.getJobId(), processConfig);
- MigrationTaskConfiguration taskConfig = buildTaskConfiguration((MigrationJobConfiguration) jobConfig, shardingItem, jobProcessContext.getPipelineProcessConfig());
- return new MigrationJobItemContext((MigrationJobConfiguration) jobConfig,
- shardingItem, (TransmissionJobItemProgress) jobItemProgress, jobProcessContext, taskConfig, getJobRunnerManager().getDataSourceManager());
+ MigrationTaskConfiguration taskConfig = buildTaskConfiguration(jobConfig, shardingItem, jobProcessContext.getPipelineProcessConfig());
+ return new MigrationJobItemContext(jobConfig, shardingItem, jobItemProgress, jobProcessContext, taskConfig, getJobRunnerManager().getDataSourceManager());
}
private MigrationTaskConfiguration buildTaskConfiguration(final MigrationJobConfiguration jobConfig, final int jobShardingItem, final PipelineProcessConfiguration processConfig) {