[GitHub] [shardingsphere] sandynz commented on a diff in pull request #19969: Add AbstractPipelineJob and move prepare method at scaling

[GitBox <gi...@apache.org>]

sandynz commented on code in PR #19969:
URL: https://github.com/apache/shardingsphere/pull/19969#discussion_r940090710


##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJob.java:
##########
@@ -17,98 +17,99 @@
 
 package org.apache.shardingsphere.data.pipeline.scenario.rulealtered;
 
-import lombok.Getter;
 import lombok.RequiredArgsConstructor;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
 import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.yaml.RuleAlteredJobConfigurationSwapper;
+import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
 import org.apache.shardingsphere.data.pipeline.api.job.PipelineJob;
 import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
+import org.apache.shardingsphere.data.pipeline.api.task.PipelineTasksRunner;
 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.datasource.PipelineDataSourceManager;
+import org.apache.shardingsphere.data.pipeline.core.exception.PipelineIgnoredException;
+import org.apache.shardingsphere.data.pipeline.core.job.PipelineJobCenter;
 import org.apache.shardingsphere.data.pipeline.core.job.progress.persist.PipelineJobProgressPersistService;
-import org.apache.shardingsphere.data.pipeline.api.task.PipelineTasksRunner;
 import org.apache.shardingsphere.elasticjob.api.ShardingContext;
-import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.OneOffJobBootstrap;
 import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
 
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-
 /**
  * Rule altered job.
  */
 @Slf4j
 @RequiredArgsConstructor
-public final class RuleAlteredJob implements SimpleJob, PipelineJob {
+public final class RuleAlteredJob extends AbstractPipelineJob implements SimpleJob, PipelineJob {
     
     private final GovernanceRepositoryAPI governanceRepositoryAPI = PipelineAPIFactory.getGovernanceRepositoryAPI();
     
-    private volatile String jobId;
-    
     private final PipelineDataSourceManager dataSourceManager = new PipelineDataSourceManager();
     
     // Shared by all sharding items
     private final RuleAlteredJobPreparer jobPreparer = new RuleAlteredJobPreparer();
     
-    @Getter
-    private final Map<Integer, PipelineTasksRunner> tasksRunnerMap = new ConcurrentHashMap<>();
-    
-    private volatile boolean stopping;
-    
-    @Setter
-    private OneOffJobBootstrap oneOffJobBootstrap;
-    
     @Override
     public void execute(final ShardingContext shardingContext) {
         log.info("Execute job {}-{}", shardingContext.getJobName(), shardingContext.getShardingItem());
-        if (stopping) {
+        if (isStopping()) {
             log.info("stopping true, ignore");
             return;
         }
-        jobId = shardingContext.getJobName();
+        setJobId(shardingContext.getJobName());
         RuleAlteredJobConfiguration jobConfig = RuleAlteredJobConfigurationSwapper.swapToObject(shardingContext.getJobParameter());
         JobProgress initProgress = governanceRepositoryAPI.getJobProgress(shardingContext.getJobName(), shardingContext.getShardingItem());
-        RuleAlteredJobContext jobContext = new RuleAlteredJobContext(jobConfig, shardingContext.getShardingItem(), initProgress, dataSourceManager, jobPreparer);
+        RuleAlteredJobContext jobContext = new RuleAlteredJobContext(jobConfig, shardingContext.getShardingItem(), initProgress, dataSourceManager);
         int shardingItem = jobContext.getShardingItem();
-        if (tasksRunnerMap.containsKey(shardingItem)) {
-            // If the following log is output, it is possible that the elasticjob task was not closed correctly
+        if (getTasksRunnerMap().containsKey(shardingItem)) {
+            // If the following log is output, it is possible that the elasticjob task was not shutdown correctly
             log.warn("schedulerMap contains shardingItem {}, ignore", shardingItem);
             return;
         }
-        log.info("start RuleAlteredJobScheduler, jobId={}, shardingItem={}", jobId, shardingItem);
+        log.info("start RuleAlteredJobScheduler, jobId={}, shardingItem={}", getJobId(), shardingItem);
         RuleAlteredJobScheduler jobScheduler = new RuleAlteredJobScheduler(jobContext);
-        jobScheduler.start();
-        tasksRunnerMap.put(shardingItem, jobScheduler);
-        PipelineJobProgressPersistService.addJobProgressPersistContext(jobId, shardingItem);
-    }
-    
-    @Override
-    public Optional<PipelineTasksRunner> getTasksRunner(final int shardingItem) {
-        return Optional.ofNullable(tasksRunnerMap.get(shardingItem));
+        runInBackground(() -> {
+            prepare(jobContext);
+            jobScheduler.start();
+        });
+        getTasksRunnerMap().put(shardingItem, jobScheduler);
+        PipelineJobProgressPersistService.addJobProgressPersistContext(getJobId(), shardingItem);
     }
     
     /**
      * Stop job.
      */
     public void stop() {
-        stopping = true;
+        setStopping(true);
         dataSourceManager.close();
-        if (null != oneOffJobBootstrap) {
-            oneOffJobBootstrap.shutdown();
+        if (null != getOneOffJobBootstrap()) {
+            getOneOffJobBootstrap().shutdown();
         }
-        if (null == jobId) {
+        if (null == getJobId()) {
             log.info("stop, jobId is null, ignore");
             return;
         }
-        log.info("stop job scheduler, jobId={}", jobId);
-        for (PipelineTasksRunner each : tasksRunnerMap.values()) {
+        log.info("stop job scheduler, jobId={}", getJobId());
+        for (PipelineTasksRunner each : getTasksRunnerMap().values()) {
             each.stop();
         }
-        tasksRunnerMap.clear();
-        PipelineJobProgressPersistService.removeJobProgressPersistContext(jobId);
+        getTasksRunnerMap().clear();
+        PipelineJobProgressPersistService.removeJobProgressPersistContext(getJobId());
+    }
+    
+    private void prepare(final RuleAlteredJobContext jobContext) {
+        try {

Review Comment:
   It's better to put `prepare` method near where it's called.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org