sandynz commented on a change in pull request #15408:
URL: https://github.com/apache/shardingsphere/pull/15408#discussion_r807953919
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
##########
@@ -314,19 +315,24 @@ public void switchClusterConfiguration(final
JobConfiguration jobConfig) {
throw new PipelineDataConsistencyCheckFailedException("Data
consistency check not finished or failed.");
}
}
- Optional<Collection<RuleAlteredJobContext>> optionalJobContexts =
RuleAlteredJobSchedulerCenter.getJobContexts(jobId);
- optionalJobContexts.ifPresent(jobContexts -> jobContexts.forEach(each
-> each.setStatus(JobStatus.ALMOST_FINISHED)));
YamlRootConfiguration yamlRootConfig =
YamlEngine.unmarshal(jobConfig.getPipelineConfig().getTarget().getParameter(),
YamlRootConfiguration.class, true);
WorkflowConfiguration workflowConfig = jobConfig.getWorkflowConfig();
String schemaName = workflowConfig.getSchemaName();
String ruleCacheId = workflowConfig.getRuleCacheId();
ScalingTaskFinishedEvent taskFinishedEvent = new
ScalingTaskFinishedEvent(schemaName, yamlRootConfig, ruleCacheId);
ShardingSphereEventBus.getInstance().post(taskFinishedEvent);
- optionalJobContexts.ifPresent(jobContexts -> jobContexts.forEach(each
-> {
- each.setStatus(JobStatus.FINISHED);
- RuleAlteredJobSchedulerCenter.persistJobProgress(each);
- }));
+
PipelineAPIFactory.getGovernanceRepositoryAPI().renewJobStatus(JobStatus.FINISHED,
jobId);
stop(jobId);
+ // TODO clean up should be done after the task is complete.
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
Review comment:
Use `log` instead of `printStackTrace`; Use `ex` install of `e`; Use
`final` for exception variable.
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
##########
@@ -314,19 +315,24 @@ public void switchClusterConfiguration(final
JobConfiguration jobConfig) {
throw new PipelineDataConsistencyCheckFailedException("Data
consistency check not finished or failed.");
}
}
- Optional<Collection<RuleAlteredJobContext>> optionalJobContexts =
RuleAlteredJobSchedulerCenter.getJobContexts(jobId);
- optionalJobContexts.ifPresent(jobContexts -> jobContexts.forEach(each
-> each.setStatus(JobStatus.ALMOST_FINISHED)));
Review comment:
`ALMOST_FINISHED` could be removed in doc, global search in project.
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobScheduler.java
##########
@@ -69,6 +63,17 @@ public void stop() {
each.stop();
each.close();
}
+ // TODO clean up should be done after the task is complete.
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ log.info("almost finished, preparer cleanup, job {}",
jobContext.getJobId());
+ RuleAlteredJobPreparer jobPreparer = jobContext.getJobPreparer();
+ if (null != jobPreparer) {
+ jobPreparer.cleanup(jobContext);
+ }
Review comment:
`cleanup` should be called only when job done and successful.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
