You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2023/12/09 14:47:52 UTC
(shardingsphere) branch master updated: Remove usage of AbstractPipelineJob.jobType (#29343)
This is an automated email from the ASF dual-hosted git repository.
zhaojinchao 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 85d38e41078 Remove usage of AbstractPipelineJob.jobType (#29343)
85d38e41078 is described below
commit 85d38e410788e014b8837d39ec83e37ceb7977a2
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sat Dec 9 22:47:46 2023 +0800
Remove usage of AbstractPipelineJob.jobType (#29343)
* Refactor GenericSQLRewriteEngineTest
* Remove useless of AbstractPipelineJob.jobType
---
.../engine/GenericSQLRewriteEngineTest.java | 2 +-
.../core/job/AbstractInseparablePipelineJob.java | 22 +++++++++++-----------
.../core/job/AbstractSeparablePipelineJob.java | 6 +++---
.../shardingsphere/data/pipeline/cdc/CDCJob.java | 4 ++--
.../consistencycheck/ConsistencyCheckJob.java | 4 +---
5 files changed, 18 insertions(+), 20 deletions(-)
diff --git a/infra/rewrite/src/test/java/org/apache/shardingsphere/infra/rewrite/engine/GenericSQLRewriteEngineTest.java b/infra/rewrite/src/test/java/org/apache/shardingsphere/infra/rewrite/engine/GenericSQLRewriteEngineTest.java
index a753669d530..fa1cff3f0e9 100644
--- a/infra/rewrite/src/test/java/org/apache/shardingsphere/infra/rewrite/engine/GenericSQLRewriteEngineTest.java
+++ b/infra/rewrite/src/test/java/org/apache/shardingsphere/infra/rewrite/engine/GenericSQLRewriteEngineTest.java
@@ -54,7 +54,7 @@ class GenericSQLRewriteEngineTest {
when(database.getResourceMetaData().getStorageUnits()).thenReturn(storageUnits);
CommonSQLStatementContext sqlStatementContext = mock(CommonSQLStatementContext.class);
when(sqlStatementContext.getDatabaseType()).thenReturn(databaseType);
- QueryContext queryContext = mock(QueryContext.class);
+ QueryContext queryContext = mock(QueryContext.class, RETURNS_DEEP_STUBS);
when(queryContext.getSqlStatementContext()).thenReturn(sqlStatementContext);
GenericSQLRewriteResult actual = new GenericSQLRewriteEngine(rule, database, mock(RuleMetaData.class))
.rewrite(new SQLRewriteContext(database, sqlStatementContext, "SELECT 1", Collections.emptyList(), mock(ConnectionContext.class),
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractInseparablePipelineJob.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractInseparablePipelineJob.java
index 9dcde50e8e2..1566479ee33 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractInseparablePipelineJob.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractInseparablePipelineJob.java
@@ -26,6 +26,7 @@ import org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfig
import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
import org.apache.shardingsphere.data.pipeline.core.job.progress.TransmissionJobItemProgress;
import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobItemManager;
+import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
import org.apache.shardingsphere.data.pipeline.core.task.PipelineTask;
import org.apache.shardingsphere.data.pipeline.core.task.runner.PipelineTasksRunner;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
@@ -42,18 +43,16 @@ import java.util.concurrent.CompletableFuture;
@Slf4j
public abstract class AbstractInseparablePipelineJob<T extends PipelineJobItemContext> extends AbstractPipelineJob {
- private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager;
-
protected AbstractInseparablePipelineJob(final String jobId) {
super(jobId);
- jobItemManager = new PipelineJobItemManager<>(getJobType().getYamlJobItemProgressSwapper());
}
@Override
public final void execute(final ShardingContext shardingContext) {
String jobId = shardingContext.getJobName();
log.info("Execute job {}", jobId);
- PipelineJobConfiguration jobConfig = getJobType().getYamlJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
+ PipelineJobType jobType = PipelineJobIdUtils.parseJobType(jobId);
+ PipelineJobConfiguration jobConfig = jobType.getYamlJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
Collection<T> jobItemContexts = new LinkedList<>();
for (int shardingItem = 0; shardingItem < jobConfig.getJobShardingCount(); shardingItem++) {
if (isStopping()) {
@@ -73,8 +72,8 @@ public abstract class AbstractInseparablePipelineJob<T extends PipelineJobItemCo
return;
}
prepare(jobItemContexts);
- executeInventoryTasks(jobItemContexts);
- executeIncrementalTasks(jobItemContexts);
+ executeInventoryTasks(jobType, jobItemContexts);
+ executeIncrementalTasks(jobType, jobItemContexts);
}
protected abstract T buildJobItemContext(PipelineJobConfiguration jobConfig, int shardingItem);
@@ -105,10 +104,10 @@ public abstract class AbstractInseparablePipelineJob<T extends PipelineJobItemCo
protected abstract void processFailed(String jobId);
- private void executeInventoryTasks(final Collection<T> jobItemContexts) {
+ private void executeInventoryTasks(final PipelineJobType jobType, final Collection<T> jobItemContexts) {
Collection<CompletableFuture<?>> futures = new LinkedList<>();
for (T each : jobItemContexts) {
- updateJobItemStatus(each, JobStatus.EXECUTE_INVENTORY_TASK);
+ updateJobItemStatus(each, jobType, JobStatus.EXECUTE_INVENTORY_TASK);
for (PipelineTask task : ((TransmissionJobItemContext) each).getInventoryTasks()) {
if (task.getTaskProgress().getPosition() instanceof FinishedPosition) {
continue;
@@ -124,12 +123,13 @@ public abstract class AbstractInseparablePipelineJob<T extends PipelineJobItemCo
protected abstract void executeInventoryTasks(Collection<CompletableFuture<?>> futures, Collection<T> jobItemContexts);
- private void updateJobItemStatus(final T jobItemContext, final JobStatus jobStatus) {
+ private void updateJobItemStatus(final T jobItemContext, final PipelineJobType jobType, final JobStatus jobStatus) {
jobItemContext.setStatus(jobStatus);
+ PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager = new PipelineJobItemManager<>(jobType.getYamlJobItemProgressSwapper());
jobItemManager.updateStatus(jobItemContext.getJobId(), jobItemContext.getShardingItem(), jobStatus);
}
- private void executeIncrementalTasks(final Collection<T> jobItemContexts) {
+ private void executeIncrementalTasks(final PipelineJobType jobType, final Collection<T> jobItemContexts) {
log.info("Execute incremental tasks, jobId={}", getJobId());
Collection<CompletableFuture<?>> futures = new LinkedList<>();
for (T each : jobItemContexts) {
@@ -137,7 +137,7 @@ public abstract class AbstractInseparablePipelineJob<T extends PipelineJobItemCo
log.info("job status already EXECUTE_INCREMENTAL_TASK, ignore");
return;
}
- updateJobItemStatus(each, JobStatus.EXECUTE_INCREMENTAL_TASK);
+ updateJobItemStatus(each, jobType, JobStatus.EXECUTE_INCREMENTAL_TASK);
for (PipelineTask task : ((TransmissionJobItemContext) each).getIncrementalTasks()) {
if (task.getTaskProgress().getPosition() instanceof FinishedPosition) {
continue;
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 ac5e44b1fe0..2c83e033062 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
@@ -55,7 +55,7 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
// CHECKSTYLE:OFF
} catch (final RuntimeException ex) {
// CHECKSTYLE:ON
- processFailed(new PipelineJobManager(getJobType()), jobId, shardingItem, ex);
+ processFailed(jobId, shardingItem, ex);
throw ex;
}
}
@@ -89,11 +89,11 @@ public abstract class AbstractSeparablePipelineJob<T extends PipelineJobItemCont
protected abstract void doPrepare(T jobItemContext) throws SQLException;
- private void processFailed(final PipelineJobManager jobManager, final String jobId, final int shardingItem, final Exception ex) {
+ private void processFailed(final String jobId, final int shardingItem, final Exception ex) {
log.error("Job execution failed, {}-{}", jobId, shardingItem, ex);
PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemFacade().getErrorMessage().update(jobId, shardingItem, ex);
try {
- jobManager.stop(jobId);
+ new PipelineJobManager(PipelineJobIdUtils.parseJobType(jobId)).stop(jobId);
} catch (final PipelineJobNotFoundException ignored) {
}
}
diff --git a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
index a9854bfd848..133ac166992 100644
--- a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
+++ b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
@@ -91,7 +91,7 @@ public final class CDCJob extends AbstractInseparablePipelineJob<CDCJobItemConte
super(jobId);
this.sink = sink;
jobAPI = (CDCJobAPI) TypedSPILoader.getService(TransmissionJobAPI.class, "STREAMING");
- jobItemManager = new PipelineJobItemManager<>(getJobType().getYamlJobItemProgressSwapper());
+ jobItemManager = new PipelineJobItemManager<>(new CDCJobType().getYamlJobItemProgressSwapper());
processConfigPersistService = new PipelineProcessConfigurationPersistService();
dataSourceManager = new DefaultPipelineDataSourceManager();
jobPreparer = new CDCJobPreparer();
@@ -101,7 +101,7 @@ public final class CDCJob extends AbstractInseparablePipelineJob<CDCJobItemConte
protected CDCJobItemContext buildJobItemContext(final PipelineJobConfiguration jobConfig, final int shardingItem) {
Optional<TransmissionJobItemProgress> initProgress = jobItemManager.getProgress(jobConfig.getJobId(), shardingItem);
PipelineProcessConfiguration processConfig = PipelineProcessConfigurationUtils.convertWithDefaultValue(
- processConfigPersistService.load(PipelineJobIdUtils.parseContextKey(jobConfig.getJobId()), getJobType().getType()));
+ processConfigPersistService.load(PipelineJobIdUtils.parseContextKey(jobConfig.getJobId()), "STREAMING"));
TransmissionProcessContext jobProcessContext = new TransmissionProcessContext(jobConfig.getJobId(), processConfig);
CDCTaskConfiguration taskConfig = buildTaskConfiguration((CDCJobConfiguration) jobConfig, shardingItem, jobProcessContext.getPipelineProcessConfig());
return new CDCJobItemContext((CDCJobConfiguration) jobConfig, shardingItem, initProgress.orElse(null), jobProcessContext, taskConfig, dataSourceManager, sink);
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 61aa3e775fc..df48a9815cc 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
@@ -17,7 +17,6 @@
package org.apache.shardingsphere.data.pipeline.scenario.consistencycheck;
-import lombok.extern.slf4j.Slf4j;
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.progress.ConsistencyCheckJobItemProgress;
@@ -34,7 +33,6 @@ import java.util.Optional;
/**
* Consistency check job.
*/
-@Slf4j
public final class ConsistencyCheckJob extends AbstractSeparablePipelineJob<ConsistencyCheckJobItemContext> {
public ConsistencyCheckJob(final String jobId) {
@@ -44,7 +42,7 @@ public final class ConsistencyCheckJob extends AbstractSeparablePipelineJob<Cons
@Override
public ConsistencyCheckJobItemContext buildJobItemContext(final ShardingContext shardingContext) {
ConsistencyCheckJobConfiguration jobConfig = new YamlConsistencyCheckJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
- PipelineJobItemManager<ConsistencyCheckJobItemProgress> jobItemManager = new PipelineJobItemManager<>(getJobType().getYamlJobItemProgressSwapper());
+ PipelineJobItemManager<ConsistencyCheckJobItemProgress> jobItemManager = new PipelineJobItemManager<>(new ConsistencyCheckJobType().getYamlJobItemProgressSwapper());
Optional<ConsistencyCheckJobItemProgress> jobItemProgress = jobItemManager.getProgress(jobConfig.getJobId(), shardingContext.getShardingItem());
return new ConsistencyCheckJobItemContext(jobConfig, shardingContext.getShardingItem(), JobStatus.RUNNING, jobItemProgress.orElse(null));
}