sandynz commented on a change in pull request #15408:
URL: https://github.com/apache/shardingsphere/pull/15408#discussion_r806404112
##########
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,18 +320,20 @@ 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:
`JobStatus.ALMOST_FINISHED` is used in `RuleAlteredJobScheduler.stop()`
to release resources via `jobPreparer.cleanup` for PostgreSQL and openGauss.
Related code:
```
final boolean almostFinished = jobContext.getStatus() ==
JobStatus.ALMOST_FINISHED;
if (almostFinished) {
log.info("almost finished, preparer cleanup, job {}",
jobContext.getJobId());
RuleAlteredJobPreparer jobPreparer = jobContext.getJobPreparer();
if (null != jobPreparer) {
jobPreparer.cleanup(jobContext);
}
}
```
Looks `JobStatus.ALMOST_FINISHED` could not be fetched from
`jobContext.getStatus()` any more after modification.
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
##########
@@ -80,6 +80,12 @@
private static final Map<String, DataConsistencyCheckAlgorithm>
DATA_CONSISTENCY_CHECK_ALGORITHM_MAP = new TreeMap<>(
SingletonSPIRegistry.getTypedSingletonInstancesMap(DataConsistencyCheckAlgorithm.class));
+ private static final JobProgressYamlSwapper JOB_PROGRESS_YAML_SWAPPER =
new JobProgressYamlSwapper();
+
+ private static final String SCALING_OFFSET_PLACEHOLDER = "%s/%s/offset";
+
+ private static final String SCALING_OFFSET_SHARDING_ITEM_PLACEHOLDER =
"%s/%s/offset/%d";
Review comment:
Could we put scaling offset path in `GovernanceRepositoryAPIImpl`, make
sure they're in the same place. And also related logic code.
New method could be added in `GovernanceRepositoryAPI`.
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/GovernanceRepositoryAPI.java
##########
@@ -98,4 +99,11 @@
* @param value value of data
*/
void persist(String key, String value);
+
+ /**
+ * renew job status.
Review comment:
`renew`'s first character should be uppercase in javadoc, and there
should be a new line between method description and params.
##########
File path:
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
##########
@@ -322,10 +322,7 @@ public void switchClusterConfiguration(final
JobConfiguration jobConfig) {
String ruleCacheId = workflowConfig.getRuleCacheId();
Review comment:
Looks `JobStatus.ALMOST_FINISHED` still might be not set, since
`optionalJobContexts` might be empty as discussed before.
--
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]
