You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by GitBox <gi...@apache.org> on 2022/08/08 08:19:33 UTC
[GitHub] [shardingsphere] azexcy opened a new pull request, #19969: Add AbstractPipelineJob and move prepare method at scaling
azexcy opened a new pull request, #19969:
URL: https://github.com/apache/shardingsphere/pull/19969
Fixes #19897.
Changes proposed in this pull request:
- Add AbstractPipelineJob
- Move prepare method
--
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
[GitHub] [shardingsphere] sandynz merged pull request #19969: Add AbstractPipelineJob and move prepare method at scaling
Posted by GitBox <gi...@apache.org>.
sandynz merged PR #19969:
URL: https://github.com/apache/shardingsphere/pull/19969
--
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
[GitHub] [shardingsphere] codecov-commenter commented on pull request #19969: Add AbstractPipelineJob and move prepare method at scaling
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #19969:
URL: https://github.com/apache/shardingsphere/pull/19969#issuecomment-1207898882
# [Codecov](https://codecov.io/gh/apache/shardingsphere/pull/19969?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#19969](https://codecov.io/gh/apache/shardingsphere/pull/19969?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b57fd63) into [master](https://codecov.io/gh/apache/shardingsphere/commit/83473dcd4e53b52ff18368da47e5b71b423df81d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (83473dc) will **increase** coverage by `0.00%`.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #19969 +/- ##
=========================================
Coverage 60.25% 60.25%
Complexity 2412 2412
=========================================
Files 3859 3862 +3
Lines 54715 54759 +44
Branches 7660 7659 -1
=========================================
+ Hits 32966 32995 +29
- Misses 18915 18930 +15
Partials 2834 2834
```
| [Impacted Files](https://codecov.io/gh/apache/shardingsphere/pull/19969?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...line/scenario/rulealtered/AbstractPipelineJob.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL3NjZW5hcmlvL3J1bGVhbHRlcmVkL0Fic3RyYWN0UGlwZWxpbmVKb2IuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../pipeline/scenario/rulealtered/RuleAlteredJob.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL3NjZW5hcmlvL3J1bGVhbHRlcmVkL1J1bGVBbHRlcmVkSm9iLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...ne/scenario/rulealtered/RuleAlteredJobContext.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL3NjZW5hcmlvL3J1bGVhbHRlcmVkL1J1bGVBbHRlcmVkSm9iQ29udGV4dC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [.../scenario/rulealtered/RuleAlteredJobScheduler.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL3NjZW5hcmlvL3J1bGVhbHRlcmVkL1J1bGVBbHRlcmVkSm9iU2NoZWR1bGVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...oneShardingStrategyConfigurationYamlConstruct.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUtZmVhdHVyZXMvc2hhcmRpbmdzcGhlcmUtc2hhcmRpbmcvc2hhcmRpbmdzcGhlcmUtc2hhcmRpbmctY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc2hhcmRpbmdzcGhlcmUvc2hhcmRpbmcveWFtbC9lbmdpbmUvY29uc3RydWN0L05vbmVTaGFyZGluZ1N0cmF0ZWd5Q29uZmlndXJhdGlvbllhbWxDb25zdHJ1Y3QuamF2YQ==) | `50.00% <0.00%> (-25.00%)` | :arrow_down: |
| [...metadata/type/c3p0/C3P0DataSourcePoolMetaData.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUtaW5mcmEvc2hhcmRpbmdzcGhlcmUtaW5mcmEtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9pbmZyYS9kYXRhc291cmNlL3Bvb2wvbWV0YWRhdGEvdHlwZS9jM3AwL0MzUDBEYXRhU291cmNlUG9vbE1ldGFEYXRhLmphdmE=) | `85.71% <0.00%> (ø)` | |
| [...ata/type/c3p0/C3P0DataSourcePoolFieldMetaData.java](https://codecov.io/gh/apache/shardingsphere/pull/19969/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUtaW5mcmEvc2hhcmRpbmdzcGhlcmUtaW5mcmEtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9pbmZyYS9kYXRhc291cmNlL3Bvb2wvbWV0YWRhdGEvdHlwZS9jM3AwL0MzUDBEYXRhU291cmNlUG9vbEZpZWxkTWV0YURhdGEuamF2YQ==) | `0.00% <0.00%> (ø)` | |
:mega: Codecov can now indicate which changes are the most critical in Pull Requests. [Learn more](https://about.codecov.io/product/feature/runtime-insights/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #19969: Add AbstractPipelineJob and move prepare method at scaling
Posted by 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