linghengqian commented on code in PR #2513:
URL:
https://github.com/apache/shardingsphere-elasticjob/pull/2513#discussion_r3414039758
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/SingleShardingJobFacade.java:
##########
@@ -48,111 +48,115 @@
*/
@Slf4j
public final class SingleShardingJobFacade extends AbstractJobFacade {
-
- private final ConfigurationService configService;
-
- private final ShardingService shardingService;
-
- private final ExecutionContextService executionContextService;
-
- private final ExecutionService executionService;
-
- private final FailoverService failoverService;
-
- private final Collection<ElasticJobListener> elasticJobListeners;
-
- private final JobTracingEventBus jobTracingEventBus;
-
- private final JobNodeStorage jobNodeStorage;
-
- private final InstanceService instanceService;
-
- public SingleShardingJobFacade(final CoordinatorRegistryCenter regCenter,
final String jobName, final Collection<ElasticJobListener> elasticJobListeners,
- final TracingConfiguration<?>
tracingConfig) {
- super(regCenter, jobName, elasticJobListeners, tracingConfig);
-
- configService = new ConfigurationService(regCenter, jobName);
- shardingService = new ShardingService(regCenter, jobName);
- executionContextService = new ExecutionContextService(regCenter,
jobName);
- executionService = new ExecutionService(regCenter, jobName);
- failoverService = new FailoverService(regCenter, jobName);
- this.elasticJobListeners =
elasticJobListeners.stream().sorted(Comparator.comparingInt(ElasticJobListener::order)).collect(Collectors.toList());
- this.jobTracingEventBus = null == tracingConfig ? new
JobTracingEventBus() : new JobTracingEventBus(tracingConfig);
- jobNodeStorage = new JobNodeStorage(regCenter, jobName);
- instanceService = new InstanceService(regCenter, jobName);
- }
-
- @Override
- public void registerJobCompleted(final ShardingContexts shardingContexts) {
- super.registerJobCompleted(shardingContexts);
-
- JobConfiguration jobConfig = configService.load(true);
- JobInstance jobInst =
JobRegistry.getInstance().getJobInstance(jobConfig.getJobName());
- if (null == jobInst) {
- log.warn("Error! Can't find the job instance with name:{}",
jobConfig.getJobName());
- return;
- }
- Integer nextIndex = null;
- List<JobInstance> availJobInst =
instanceService.getAvailableJobInstances();
- for (int i = 0; i < availJobInst.size(); i++) {
- JobInstance temp = availJobInst.get(i);
- if (temp.getServerIp().equals(jobInst.getServerIp())) {
- nextIndex = i + 1;
- break;
- }
- }
- if (nextIndex != null) {
- nextIndex = nextIndex >= availJobInst.size() ? 0 : nextIndex;
- jobNodeStorage.fillEphemeralJobNode("next-job-instance-ip",
availJobInst.get(nextIndex).getServerIp());
- }
-
- if (log.isDebugEnabled()) {
- log.debug("job name: {}, next index: {}, sharding total count: {}",
- jobConfig.getJobName(), nextIndex,
jobConfig.getShardingTotalCount());
- }
- }
-
- /**
- * Get sharding contexts.
- *
- * @return sharding contexts
- */
- @Override
- public ShardingContexts getShardingContexts() {
- JobConfiguration jobConfig = configService.load(true);
- boolean isFailover = jobConfig.isFailover();
- if (isFailover) {
- List<Integer> failoverShardingItems =
failoverService.getLocalFailoverItems();
- if (!failoverShardingItems.isEmpty()) {
- return
executionContextService.getJobShardingContext(failoverShardingItems);
- }
- }
-
- List<Integer> shardingItems;
- String nextJobInstIP = null;
- if (isNeedSharding()) {
- shardingService.shardingIfNecessary();
- shardingItems = shardingService.getLocalShardingItems();
- } else {
- nextJobInstIP =
jobNodeStorage.getJobNodeDataDirectly("next-job-instance-ip");
- if (StringUtils.isBlank(nextJobInstIP)) {
- shardingService.shardingIfNecessary();
- shardingItems = shardingService.getLocalShardingItems();
- } else {
- JobInstance jobInst =
JobRegistry.getInstance().getJobInstance(jobConfig.getJobName());
- shardingItems = nextJobInstIP.equals(jobInst.getServerIp()) ?
Collections.singletonList(0) : new ArrayList<>();
- }
- }
- if (log.isDebugEnabled()) {
- log.debug("job name: {}, sharding items: {}, nextJobInstIP: {},
sharding total count: {}, isFailover: {}",
- jobConfig.getJobName(), shardingItems, nextJobInstIP,
jobConfig.getShardingTotalCount(), isFailover);
- }
-
- if (isFailover) {
- shardingItems.removeAll(failoverService.getLocalTakeOffItems());
- }
-
shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
- return executionContextService.getJobShardingContext(shardingItems);
- }
-
Review Comment:
Are you using git for Windows on Windows instead of the git in wsl2? I'm
confused as to why this change was made.
In fact, the main maintainers of Elasticjob are all using Linux or WSL2, and
no one has paid attention to Git configuration under Windows native for a long
time. You can add something similar to
https://docs.github.com/en/get-started/git-basics/configuring-git-to-handle-line-endings
to the `.gitattributes` file in the project root directory, and I can review
it.
--
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]
