Skip to content

Commit 012d7de

Browse files
author
baiyangzhuhong
committed
fix problems of code formatting validation
1 parent f9296ab commit 012d7de

File tree

9 files changed

+172
-79
lines changed

9 files changed

+172
-79
lines changed

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/AbstractJobFacade.java

+16-16
Original file line numberDiff line numberDiff line change
@@ -48,22 +48,22 @@
4848
*/
4949
@Slf4j
5050
abstract class AbstractJobFacade implements JobFacade {
51-
52-
protected final ConfigurationService configService;
53-
54-
protected final ShardingService shardingService;
55-
56-
protected final ExecutionContextService executionContextService;
57-
58-
protected final ExecutionService executionService;
59-
60-
protected final FailoverService failoverService;
61-
62-
protected final Collection<ElasticJobListener> elasticJobListeners;
63-
64-
protected final JobTracingEventBus jobTracingEventBus;
65-
66-
public AbstractJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration<?> tracingConfig) {
51+
52+
private final ConfigurationService configService;
53+
54+
private final ShardingService shardingService;
55+
56+
private final ExecutionContextService executionContextService;
57+
58+
private final ExecutionService executionService;
59+
60+
private final FailoverService failoverService;
61+
62+
private final Collection<ElasticJobListener> elasticJobListeners;
63+
64+
private final JobTracingEventBus jobTracingEventBus;
65+
66+
AbstractJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration<?> tracingConfig) {
6767
configService = new ConfigurationService(regCenter, jobName);
6868
shardingService = new ShardingService(regCenter, jobName);
6969
executionContextService = new ExecutionContextService(regCenter, jobName);

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/JobFacade.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ public interface JobFacade {
3737
* @param fromCache load from cache or not
3838
* @return job configuration
3939
*/
40-
JobConfiguration loadJobConfiguration(final boolean fromCache);
40+
JobConfiguration loadJobConfiguration(boolean fromCache);
4141

4242
/**
4343
* Check job execution environment.
@@ -121,7 +121,7 @@ public interface JobFacade {
121121
*
122122
* @param jobExecutionEvent job execution event
123123
*/
124-
void postJobExecutionEvent(final JobExecutionEvent jobExecutionEvent);
124+
void postJobExecutionEvent(JobExecutionEvent jobExecutionEvent);
125125

126126
/**
127127
* Post job status trace event.
@@ -131,7 +131,7 @@ public interface JobFacade {
131131
* @param message job message
132132
*/
133133
void postJobStatusTraceEvent(String taskId, JobStatusTraceEvent.State state, String message);
134-
134+
135135
/**
136136
* Get job runtime service.
137137
*

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/ShardingJobFacade.java

+33-2
Original file line numberDiff line numberDiff line change
@@ -18,24 +18,55 @@
1818
package org.apache.shardingsphere.elasticjob.kernel.executor.facade;
1919

2020
import lombok.extern.slf4j.Slf4j;
21+
22+
import org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
23+
import org.apache.shardingsphere.elasticjob.kernel.internal.failover.FailoverService;
24+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ExecutionContextService;
25+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ExecutionService;
26+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ShardingService;
27+
import org.apache.shardingsphere.elasticjob.kernel.tracing.event.JobTracingEventBus;
2128
import org.apache.shardingsphere.elasticjob.spi.listener.ElasticJobListener;
2229
import org.apache.shardingsphere.elasticjob.spi.listener.param.ShardingContexts;
2330
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
2431
import org.apache.shardingsphere.elasticjob.kernel.tracing.config.TracingConfiguration;
2532

2633
import java.util.Collection;
34+
import java.util.Comparator;
2735
import java.util.List;
36+
import java.util.stream.Collectors;
2837

2938
/**
3039
* Sharding Job facade.
3140
*/
3241
@Slf4j
3342
public final class ShardingJobFacade extends AbstractJobFacade {
3443

44+
private final ConfigurationService configService;
45+
46+
private final ShardingService shardingService;
47+
48+
private final ExecutionContextService executionContextService;
49+
50+
private final ExecutionService executionService;
51+
52+
private final FailoverService failoverService;
53+
54+
private final Collection<ElasticJobListener> elasticJobListeners;
55+
56+
private final JobTracingEventBus jobTracingEventBus;
57+
3558
public ShardingJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration<?> tracingConfig) {
3659
super(regCenter, jobName, elasticJobListeners, tracingConfig);
60+
61+
configService = new ConfigurationService(regCenter, jobName);
62+
shardingService = new ShardingService(regCenter, jobName);
63+
executionContextService = new ExecutionContextService(regCenter, jobName);
64+
executionService = new ExecutionService(regCenter, jobName);
65+
failoverService = new FailoverService(regCenter, jobName);
66+
this.elasticJobListeners = elasticJobListeners.stream().sorted(Comparator.comparingInt(ElasticJobListener::order)).collect(Collectors.toList());
67+
this.jobTracingEventBus = null == tracingConfig ? new JobTracingEventBus() : new JobTracingEventBus(tracingConfig);
3768
}
38-
69+
3970
/**
4071
* Get sharding contexts.
4172
*
@@ -58,5 +89,5 @@ public ShardingContexts getShardingContexts() {
5889
shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
5990
return executionContextService.getJobShardingContext(shardingItems);
6091
}
61-
92+
6293
}

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/SingleShardingJobFacade.java

+50-19
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,23 @@
2020
import java.util.ArrayList;
2121
import java.util.Collection;
2222
import java.util.Collections;
23+
import java.util.Comparator;
2324
import java.util.List;
25+
import java.util.stream.Collectors;
2426

2527
import org.apache.commons.lang3.StringUtils;
2628
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
29+
import org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
30+
import org.apache.shardingsphere.elasticjob.kernel.internal.failover.FailoverService;
2731
import org.apache.shardingsphere.elasticjob.kernel.internal.instance.InstanceService;
2832
import org.apache.shardingsphere.elasticjob.kernel.internal.schedule.JobRegistry;
33+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ExecutionContextService;
34+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ExecutionService;
2935
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.JobInstance;
36+
import org.apache.shardingsphere.elasticjob.kernel.internal.sharding.ShardingService;
3037
import org.apache.shardingsphere.elasticjob.kernel.internal.storage.JobNodeStorage;
3138
import org.apache.shardingsphere.elasticjob.kernel.tracing.config.TracingConfiguration;
39+
import org.apache.shardingsphere.elasticjob.kernel.tracing.event.JobTracingEventBus;
3240
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
3341
import org.apache.shardingsphere.elasticjob.spi.listener.ElasticJobListener;
3442
import org.apache.shardingsphere.elasticjob.spi.listener.param.ShardingContexts;
@@ -40,21 +48,44 @@
4048
*/
4149
@Slf4j
4250
public final class SingleShardingJobFacade extends AbstractJobFacade {
43-
51+
52+
private final ConfigurationService configService;
53+
54+
private final ShardingService shardingService;
55+
56+
private final ExecutionContextService executionContextService;
57+
58+
private final ExecutionService executionService;
59+
60+
private final FailoverService failoverService;
61+
62+
private final Collection<ElasticJobListener> elasticJobListeners;
63+
64+
private final JobTracingEventBus jobTracingEventBus;
65+
4466
private final JobNodeStorage jobNodeStorage;
67+
4568
private final InstanceService instanceService;
46-
47-
public SingleShardingJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners, final TracingConfiguration<?> tracingConfig) {
69+
70+
public SingleShardingJobFacade(final CoordinatorRegistryCenter regCenter, final String jobName, final Collection<ElasticJobListener> elasticJobListeners,
71+
final TracingConfiguration<?> tracingConfig) {
4872
super(regCenter, jobName, elasticJobListeners, tracingConfig);
49-
73+
74+
configService = new ConfigurationService(regCenter, jobName);
75+
shardingService = new ShardingService(regCenter, jobName);
76+
executionContextService = new ExecutionContextService(regCenter, jobName);
77+
executionService = new ExecutionService(regCenter, jobName);
78+
failoverService = new FailoverService(regCenter, jobName);
79+
this.elasticJobListeners = elasticJobListeners.stream().sorted(Comparator.comparingInt(ElasticJobListener::order)).collect(Collectors.toList());
80+
this.jobTracingEventBus = null == tracingConfig ? new JobTracingEventBus() : new JobTracingEventBus(tracingConfig);
5081
jobNodeStorage = new JobNodeStorage(regCenter, jobName);
5182
instanceService = new InstanceService(regCenter, jobName);
5283
}
53-
84+
5485
@Override
5586
public void registerJobCompleted(final ShardingContexts shardingContexts) {
5687
super.registerJobCompleted(shardingContexts);
57-
88+
5889
JobConfiguration jobConfig = configService.load(true);
5990
JobInstance jobInst = JobRegistry.getInstance().getJobInstance(jobConfig.getJobName());
6091
if (null == jobInst) {
@@ -66,21 +97,21 @@ public void registerJobCompleted(final ShardingContexts shardingContexts) {
6697
for (int i = 0; i < availJobInst.size(); i++) {
6798
JobInstance temp = availJobInst.get(i);
6899
if (temp.getServerIp().equals(jobInst.getServerIp())) {
69-
nextIndex = i + 1; // find the current running job instance, and set next one to current index + 1
100+
nextIndex = i + 1;
70101
break;
71102
}
72103
}
73-
if (nextIndex != null) { // the normal case that can find the next index, exclude the bounded scenarios
74-
nextIndex = nextIndex >= availJobInst.size() ? 0 : nextIndex; // Round Robin Loop
104+
if (nextIndex != null) {
105+
nextIndex = nextIndex >= availJobInst.size() ? 0 : nextIndex;
75106
jobNodeStorage.fillEphemeralJobNode("next-job-instance-ip", availJobInst.get(nextIndex).getServerIp());
76107
}
77-
108+
78109
if (log.isDebugEnabled()) {
79110
log.debug("job name: {}, next index: {}, sharding total count: {}",
80-
jobConfig.getJobName(), nextIndex, jobConfig.getShardingTotalCount());
111+
jobConfig.getJobName(), nextIndex, jobConfig.getShardingTotalCount());
81112
}
82113
}
83-
114+
84115
/**
85116
* Get sharding contexts.
86117
*
@@ -96,32 +127,32 @@ public ShardingContexts getShardingContexts() {
96127
return executionContextService.getJobShardingContext(failoverShardingItems);
97128
}
98129
}
99-
130+
100131
List<Integer> shardingItems;
101132
String nextJobInstIP = null;
102-
if (isNeedSharding()) { // the first initialization or reconcile case
133+
if (isNeedSharding()) {
103134
shardingService.shardingIfNecessary();
104135
shardingItems = shardingService.getLocalShardingItems();
105136
} else {
106137
nextJobInstIP = jobNodeStorage.getJobNodeDataDirectly("next-job-instance-ip");
107-
if (StringUtils.isBlank(nextJobInstIP)) { // if there is no next job instance ip
138+
if (StringUtils.isBlank(nextJobInstIP)) {
108139
shardingService.shardingIfNecessary();
109140
shardingItems = shardingService.getLocalShardingItems();
110-
} else { // when next job instance is specified under normal case
141+
} else {
111142
JobInstance jobInst = JobRegistry.getInstance().getJobInstance(jobConfig.getJobName());
112143
shardingItems = nextJobInstIP.equals(jobInst.getServerIp()) ? Collections.singletonList(0) : new ArrayList<>();
113144
}
114145
}
115146
if (log.isDebugEnabled()) {
116147
log.debug("job name: {}, sharding items: {}, nextJobInstIP: {}, sharding total count: {}, isFailover: {}",
117-
jobConfig.getJobName(), shardingItems, nextJobInstIP, jobConfig.getShardingTotalCount(), isFailover);
148+
jobConfig.getJobName(), shardingItems, nextJobInstIP, jobConfig.getShardingTotalCount(), isFailover);
118149
}
119-
150+
120151
if (isFailover) {
121152
shardingItems.removeAll(failoverService.getLocalTakeOffItems());
122153
}
123154
shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
124155
return executionContextService.getJobShardingContext(shardingItems);
125156
}
126-
157+
127158
}

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/schedule/JobScheduler.java

+10-8
Original file line numberDiff line numberDiff line change
@@ -82,14 +82,15 @@ public JobScheduler(final CoordinatorRegistryCenter regCenter, final ElasticJob
8282
Collection<ElasticJobListener> jobListeners = getElasticJobListeners(this.jobConfig);
8383
setUpFacade = new SetUpFacade(regCenter, this.jobConfig.getJobName(), jobListeners);
8484
schedulerFacade = new SchedulerFacade(regCenter, this.jobConfig.getJobName());
85-
86-
if (1 == this.jobConfig.getShardingTotalCount() // the single sharding scenario
87-
&& "SINGLE_SHARDING_BALANCE".equals(this.jobConfig.getJobShardingStrategyType())) { // the specified SINGLE_SHARDING_BALANCE strategy
85+
86+
// the single sharding scenario and specified SINGLE_SHARDING_BALANCE strategy
87+
if (1 == this.jobConfig.getShardingTotalCount()
88+
&& "SINGLE_SHARDING_BALANCE".equals(this.jobConfig.getJobShardingStrategyType())) {
8889
jobFacade = new SingleShardingJobFacade(regCenter, this.jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null));
8990
} else {
9091
jobFacade = new ShardingJobFacade(regCenter, this.jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null));
9192
}
92-
93+
9394
validateJobProperties();
9495
jobExecutor = new ElasticJobExecutor(elasticJob, this.jobConfig, jobFacade);
9596
setGuaranteeServiceForElasticJobListeners(regCenter, jobListeners);
@@ -103,14 +104,15 @@ public JobScheduler(final CoordinatorRegistryCenter regCenter, final String elas
103104
Collection<ElasticJobListener> jobListeners = getElasticJobListeners(this.jobConfig);
104105
setUpFacade = new SetUpFacade(regCenter, this.jobConfig.getJobName(), jobListeners);
105106
schedulerFacade = new SchedulerFacade(regCenter, this.jobConfig.getJobName());
106-
107-
if (1 == this.jobConfig.getShardingTotalCount() // the single sharding scenario
108-
&& "SINGLE_SHARDING_BALANCE".equals(this.jobConfig.getJobShardingStrategyType())) { // the specified SINGLE_SHARDING_BALANCE strategy
107+
108+
// the single sharding scenario and specified SINGLE_SHARDING_BALANCE strategy
109+
if (1 == this.jobConfig.getShardingTotalCount()
110+
&& "SINGLE_SHARDING_BALANCE".equals(this.jobConfig.getJobShardingStrategyType())) {
109111
jobFacade = new SingleShardingJobFacade(regCenter, this.jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null));
110112
} else {
111113
jobFacade = new ShardingJobFacade(regCenter, this.jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null));
112114
}
113-
115+
114116
validateJobProperties();
115117
jobExecutor = new ElasticJobExecutor(elasticJobType, this.jobConfig, jobFacade);
116118
setGuaranteeServiceForElasticJobListeners(regCenter, jobListeners);

‎kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/strategy/type/SingleShardingBalanceJobShardingStrategy.java

+24-7
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,20 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
118
package org.apache.shardingsphere.elasticjob.kernel.internal.sharding.strategy.type;
219

320
import java.util.ArrayList;
@@ -22,26 +39,26 @@
2239
* @since 2024-12-03 19:19
2340
*/
2441
public class SingleShardingBalanceJobShardingStrategy implements JobShardingStrategy {
25-
42+
2643
private final AverageAllocationJobShardingStrategy averageAllocationJobShardingStrategy = new AverageAllocationJobShardingStrategy();
27-
44+
2845
@Override
2946
public Map<JobInstance, List<Integer>> sharding(final List<JobInstance> jobInstances, final String jobName, final int shardingTotalCount) {
3047
int shardingUnitsSize = jobInstances.size();
31-
int offset = Math.abs(jobName.hashCode() + ((Long)System.currentTimeMillis()).intValue()) % shardingUnitsSize;
32-
48+
int offset = Math.abs(jobName.hashCode() + ((Long) System.currentTimeMillis()).intValue()) % shardingUnitsSize;
49+
3350
List<JobInstance> result = new ArrayList<>(shardingUnitsSize);
3451
for (int i = 0; i < shardingUnitsSize; i++) {
3552
int index = (i + offset) % shardingUnitsSize;
3653
result.add(jobInstances.get(index));
3754
}
38-
55+
3956
return averageAllocationJobShardingStrategy.sharding(result, jobName, shardingTotalCount);
4057
}
41-
58+
4259
@Override
4360
public String getType() {
4461
return "SINGLE_SHARDING_BALANCE";
4562
}
46-
63+
4764
}

‎kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/ShardingJobFacadeTest.java

+6
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,12 @@ void setUp() {
8484
ReflectionUtils.setSuperclassFieldValue(shardingJobFacade, "executionService", executionService);
8585
ReflectionUtils.setSuperclassFieldValue(shardingJobFacade, "failoverService", failoverService);
8686
ReflectionUtils.setSuperclassFieldValue(shardingJobFacade, "jobTracingEventBus", jobTracingEventBus);
87+
ReflectionUtils.setFieldValue(shardingJobFacade, "configService", configService);
88+
ReflectionUtils.setFieldValue(shardingJobFacade, "shardingService", shardingService);
89+
ReflectionUtils.setFieldValue(shardingJobFacade, "executionContextService", executionContextService);
90+
ReflectionUtils.setFieldValue(shardingJobFacade, "executionService", executionService);
91+
ReflectionUtils.setFieldValue(shardingJobFacade, "failoverService", failoverService);
92+
ReflectionUtils.setFieldValue(shardingJobFacade, "jobTracingEventBus", jobTracingEventBus);
8793
}
8894

8995
@Test

0 commit comments

Comments
 (0)