dahyvuun commented on code in PR #2513:
URL:
https://github.com/apache/shardingsphere-elasticjob/pull/2513#discussion_r3414076936
##########
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:
Yes, I'm using IntelliJ IDEA on Windows. I've added a .gitattributes file
with * text=auto eol=lf to enforce LF line endings and renormalized existing
files. Thank you for the guidance!
--
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]
