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 2022/04/13 09:44:48 UTC
[shardingsphere] branch master updated: Improve scaling job status persistence (#16791)
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 5a6b84e82b7 Improve scaling job status persistence (#16791)
5a6b84e82b7 is described below
commit 5a6b84e82b7b8aff3cc54af9f5e3b5c8160497ef
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Wed Apr 13 17:44:41 2022 +0800
Improve scaling job status persistence (#16791)
---
.../pipeline/core/api/impl/RuleAlteredJobAPIImpl.java | 4 +++-
.../rulealtered/RuleAlteredJobSchedulerCenter.java | 18 ++++++++++++++++++
2 files changed, 21 insertions(+), 1 deletion(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
index 370265c5dea..55d012cc69c 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
@@ -356,6 +356,8 @@ public final class RuleAlteredJobAPIImpl extends AbstractPipelineJobAPIImpl impl
WorkflowConfiguration workflowConfig = jobConfig.getWorkflowConfig();
ScalingTaskFinishedEvent taskFinishedEvent = new ScalingTaskFinishedEvent(workflowConfig.getSchemaName(), workflowConfig.getActiveVersion(), workflowConfig.getNewVersion());
ShardingSphereEventBus.getInstance().post(taskFinishedEvent);
+ // TODO rewrite job status update after job progress structure refactor
+ RuleAlteredJobSchedulerCenter.updateJobStatus(jobId, JobStatus.FINISHED);
for (int each : repositoryAPI.getShardingItems(jobId)) {
repositoryAPI.updateShardingJobStatus(jobId, each, JobStatus.FINISHED);
}
@@ -372,7 +374,7 @@ public final class RuleAlteredJobAPIImpl extends AbstractPipelineJobAPIImpl impl
try {
new ScalingEnvironmentManager().cleanupTargetTables(getJobConfig(jobConfigPOJO));
} catch (final SQLException ex) {
- throw new PipelineJobExecutionException("Reset target table failed for job " + jobId);
+ throw new PipelineJobExecutionException("Reset target table failed for job " + jobId, ex);
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobSchedulerCenter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobSchedulerCenter.java
index aae0bbfff89..5b01a702aad 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobSchedulerCenter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobSchedulerCenter.java
@@ -21,6 +21,7 @@ import com.google.common.collect.Maps;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory;
import org.apache.shardingsphere.infra.executor.kernel.thread.ExecutorThreadFactoryBuilder;
@@ -93,6 +94,23 @@ public final class RuleAlteredJobSchedulerCenter {
return JOB_SCHEDULER_MAP.containsKey(jobId);
}
+ /**
+ * Update job status for all job sharding.
+ *
+ * @param jobId job id
+ * @param jobStatus job status
+ */
+ public static void updateJobStatus(final String jobId, final JobStatus jobStatus) {
+ Map<Integer, RuleAlteredJobScheduler> schedulerMap = JOB_SCHEDULER_MAP.get(jobId);
+ if (null == schedulerMap) {
+ log.info("updateJobStatus, schedulerMap is null, ignore");
+ return;
+ }
+ for (Entry<Integer, RuleAlteredJobScheduler> entry : schedulerMap.entrySet()) {
+ entry.getValue().getJobContext().setStatus(jobStatus);
+ }
+ }
+
private static final class PersistJobContextRunnable implements Runnable {
@Override