This is an automated email from the ASF dual-hosted git repository.
linghengqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere-elasticjob.git
The following commit(s) were added to refs/heads/master by this push:
new 6697fd0e8 Fix indefinite blocking in sharding, leader election, and
distributed-once listener wait loops (#2546)
6697fd0e8 is described below
commit 6697fd0e87f8c15c63872b55cd59bc4da7d4f00e
Author: conellimo <[email protected]>
AuthorDate: Tue Oct 13 07:14:16 2026 +0300
Fix indefinite blocking in sharding, leader election, and distributed-once
listener wait loops (#2546)
---
RELEASE-NOTES.md | 2 +
.../annotation/ElasticJobConfiguration.java | 7 ++
.../elasticjob/api/JobConfiguration.java | 22 ++++-
.../content/user-manual/configuration/_index.cn.md | 5 +
.../content/user-manual/configuration/_index.en.md | 5 +
.../user-manual/configuration/java-api.cn.md | 1 +
.../user-manual/configuration/java-api.en.md | 1 +
.../configuration/spring-boot-starter.cn.md | 1 +
.../configuration/spring-boot-starter.en.md | 1 +
.../configuration/spring-namespace.cn.md | 1 +
.../configuration/spring-namespace.en.md | 1 +
.../kernel/infra/util/BoundedWaitGuard.java | 65 +++++++++++++
.../kernel/infra/util/WaitTimeoutNotifier.java | 56 +++++++++++
.../internal/annotation/JobAnnotationBuilder.java | 3 +-
.../internal/config/JobConfigurationPOJO.java | 5 +-
.../kernel/internal/election/LeaderService.java | 26 ++++-
.../kernel/internal/sharding/ShardingService.java | 38 +++++++-
.../kernel/internal/storage/JobNodeStorage.java | 25 +++++
.../AbstractDistributeOnceElasticJobListener.java | 4 +-
.../JobAnnotationBuilderMaxWaitMillisTest.java | 44 +++++++++
.../JobConfigurationPOJOMaxWaitMillisTest.java | 37 +++++++
.../internal/config/JobConfigurationPOJOTest.java | 2 +
.../election/LeaderServiceBoundedWaitTest.java | 95 ++++++++++++++++++
.../internal/election/LeaderServiceTest.java | 7 ++
.../sharding/ShardingServiceBoundedWaitTest.java | 108 +++++++++++++++++++++
.../internal/sharding/ShardingServiceTest.java | 20 +++-
...ibuteOnceElasticJobListenerBoundedWaitTest.java | 92 ++++++++++++++++++
.../lifecycle/fixture/LifecycleYamlConstants.java | 1 +
.../job/ElasticJobConfigurationProperties.java | 4 +-
...obConfigurationPropertiesMaxWaitMillisTest.java | 47 +++++++++
.../job/parser/JobBeanDefinitionParser.java | 1 +
.../namespace/job/tag/JobBeanDefinitionTag.java | 2 +
.../resources/META-INF/namespace/elasticjob.xsd | 1 +
.../JobBeanDefinitionParserMaxWaitMillisTest.java | 31 ++++++
34 files changed, 750 insertions(+), 11 deletions(-)
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 738adf7fd..374da0609 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -3,6 +3,7 @@
### API Changes
1. Spring Boot Starter: Rename `ZookeeperProperties` to
`RegistryCenterProperties`, the registry center type is selected by the new
`elasticjob.reg-center.type` property -
[#2554](https://github.com/apache/shardingsphere-elasticjob/pull/2554)
+1. Kernel: Adds the `maxWaitMillis` option, which bounds how long the sharding
and leader election wait loops block, to `JobConfiguration` and to the YAML,
Spring Boot, Spring Namespace and annotation configurations -
[#2546](https://github.com/apache/shardingsphere-elasticjob/pull/2546)
### Enhancements
@@ -17,6 +18,7 @@
1. Kernel: Fixes the issue that the distributed once listener loops forever
when the registration is not confirmed by the registry center -
[#2543](https://github.com/apache/shardingsphere-elasticjob/issues/2543)
1. Kernel: Fixes the issue that the trigger flag is removed before the job is
triggered -
[#2543](https://github.com/apache/shardingsphere-elasticjob/issues/2543)
1. Kernel: Fixes the issues that a job keeps running or throws an exception
after it is shut down or its configuration is removed -
[#2549](https://github.com/apache/shardingsphere-elasticjob/issues/2549)
+1. Kernel: Fixes the issue that the sharding and leader election wait loops
block indefinitely when the registry center never clears the awaited state -
[#2546](https://github.com/apache/shardingsphere-elasticjob/pull/2546)
1. Registry Center: Fixes the issue that the etcd registry center returns
incorrect child keys under a nested path -
[#2538](https://github.com/apache/shardingsphere-elasticjob/issues/2538)
1. Registry Center: Fixes the issue that the etcd registry center does not
report the root path as existed -
[#2538](https://github.com/apache/shardingsphere-elasticjob/issues/2538)
1. Registry Center: Fixes the issue that the etcd registry center does not
revoke the lease created by leader execution -
[#2538](https://github.com/apache/shardingsphere-elasticjob/issues/2538)
diff --git
a/api/src/main/java/org/apache/shardingsphere/elasticjob/annotation/ElasticJobConfiguration.java
b/api/src/main/java/org/apache/shardingsphere/elasticjob/annotation/ElasticJobConfiguration.java
index b9ddc3211..23a09c430 100644
---
a/api/src/main/java/org/apache/shardingsphere/elasticjob/annotation/ElasticJobConfiguration.java
+++
b/api/src/main/java/org/apache/shardingsphere/elasticjob/annotation/ElasticJobConfiguration.java
@@ -179,4 +179,11 @@ public @interface ElasticJobConfiguration {
* @return overwrite
*/
boolean overwrite() default false;
+
+ /**
+ * Maximum time in milliseconds that the coordination wait loops, such as
sharding and leader election, block before giving up.
+ *
+ * @return max wait milliseconds
+ */
+ long maxWaitMillis() default 60_000L;
}
diff --git
a/api/src/main/java/org/apache/shardingsphere/elasticjob/api/JobConfiguration.java
b/api/src/main/java/org/apache/shardingsphere/elasticjob/api/JobConfiguration.java
index c48dee1e5..a940d53d5 100644
---
a/api/src/main/java/org/apache/shardingsphere/elasticjob/api/JobConfiguration.java
+++
b/api/src/main/java/org/apache/shardingsphere/elasticjob/api/JobConfiguration.java
@@ -81,6 +81,8 @@ public final class JobConfiguration {
private final boolean staticSharding;
+ private final long maxWaitMillis;
+
/**
* Create ElasticJob configuration builder.
*
@@ -139,6 +141,8 @@ public final class JobConfiguration {
private boolean staticSharding;
+ private long maxWaitMillis = 60_000L;
+
/**
* Cron expression.
*
@@ -411,6 +415,22 @@ public final class JobConfiguration {
return this;
}
+ /**
+ * Set max wait milliseconds for coordination wait loops (e.g.
sharding, leader election) to give up instead of blocking indefinitely.
+ *
+ * <p>
+ * Each wait is bounded separately. When the limit elapses or the
waiting thread is interrupted, the wait stops, the situation is reported to the
job error handler,
+ * and the caller continues without the awaited state having been
reached.
+ * </p>
+ *
+ * @param maxWaitMillis max wait milliseconds
+ * @return ElasticJob configuration builder
+ */
+ public Builder maxWaitMillis(final long maxWaitMillis) {
+ this.maxWaitMillis = maxWaitMillis;
+ return this;
+ }
+
/**
* Build ElasticJob configuration.
*
@@ -422,7 +442,7 @@ public final class JobConfiguration {
return new JobConfiguration(jobName, cron, timeZone,
shardingTotalCount, shardingItemParameters, jobParameter,
monitorExecution, failover, misfire, maxTimeDiffSeconds,
reconcileIntervalMinutes,
jobShardingStrategyType,
jobExecutorThreadPoolSizeProviderType, jobErrorHandlerType, jobListenerTypes,
- extraConfigurations, description, props, disabled,
overwrite, label, staticSharding);
+ extraConfigurations, description, props, disabled,
overwrite, label, staticSharding, maxWaitMillis);
}
}
}
diff --git a/docs/content/user-manual/configuration/_index.cn.md
b/docs/content/user-manual/configuration/_index.cn.md
index ea5517611..e77f04ceb 100644
--- a/docs/content/user-manual/configuration/_index.cn.md
+++ b/docs/content/user-manual/configuration/_index.cn.md
@@ -63,6 +63,7 @@ ElasticJob 支持多种注册中心类型，详细配置请参见[注册中心
| props | Properties | | 作业属性配置信息
|
| disabled | boolean | false | 作业是否禁止启动
|
| overwrite | boolean | false |
本地配置是否可覆盖注册中心配置 |
+| maxWaitMillis | long | 60000 |
分片、主节点选举等协调等待循环放弃等待前的最长阻塞时间（毫秒） |
### 核心配置项说明
@@ -118,6 +119,10 @@ ElasticJob 支持多种注册中心类型，详细配置请参见[注册中心
如果可覆盖，每次启动作业都以本地配置为准。
+**maxWaitMillis:**
+
+分片、主节点选举等协调等待循环在等待注册中心状态变化时的最长阻塞时间，单位毫秒，每次等待单独计算。超时或等待线程被中断后，将停止等待，并通过作业错误处理器上报，作业继续执行。默认值为
`60000`。
+
## 作业监听器配置项
### 常规监听器配置项
diff --git a/docs/content/user-manual/configuration/_index.en.md
b/docs/content/user-manual/configuration/_index.en.md
index 16f648894..3c00535aa 100644
--- a/docs/content/user-manual/configuration/_index.en.md
+++ b/docs/content/user-manual/configuration/_index.en.md
@@ -62,6 +62,7 @@ Include IP and port, multiple addresses are separated by
commas.
| props | Properties | | Job
properties
|
| disabled | boolean | false | Enable or
disable start the job |
| overwrite | boolean | false | Enable or
disable local configuration override registry center configuration |
+| maxWaitMillis | long | 60000 | Maximum
time in milliseconds that the sharding and leader election wait loops block
before giving up |
### Core Configuration Description
@@ -117,6 +118,10 @@ It can be used for deployment, forbid jobs to start, and
then start them uniform
If the value is `true`, local configuration override registry center
configuration every time the job is started.
+**maxWaitMillis:**
+
+The maximum time in milliseconds that the sharding and leader election wait
loops block while waiting for the registry center state to change. The limit
applies to each wait separately. When it elapses, or the waiting thread is
interrupted, the wait stops, the situation is reported to the job error
handler, and the job continues. Defaults to `60000`.
+
## Job Listener Configuration
### Common Listener Configuration
diff --git a/docs/content/user-manual/configuration/java-api.cn.md
b/docs/content/user-manual/configuration/java-api.cn.md
index 5e1af30db..8e4943ca5 100644
--- a/docs/content/user-manual/configuration/java-api.cn.md
+++ b/docs/content/user-manual/configuration/java-api.cn.md
@@ -64,3 +64,4 @@ ElasticJob 支持 `ZooKeeper`（默认）、`etcd`、`Nacos` 和内存注册中
| props | 否 |
| disabled | 否 |
| overwrite | 否 |
+| maxWaitMillis | 否 |
diff --git a/docs/content/user-manual/configuration/java-api.en.md
b/docs/content/user-manual/configuration/java-api.en.md
index 7ad0c7a4d..1f389d773 100644
--- a/docs/content/user-manual/configuration/java-api.en.md
+++ b/docs/content/user-manual/configuration/java-api.en.md
@@ -65,3 +65,4 @@ Configuration:
| props | No |
| disabled | No |
| overwrite | No |
+| maxWaitMillis | No |
diff --git a/docs/content/user-manual/configuration/spring-boot-starter.cn.md
b/docs/content/user-manual/configuration/spring-boot-starter.cn.md
index dd7a250d4..0e3b92fc9 100644
--- a/docs/content/user-manual/configuration/spring-boot-starter.cn.md
+++ b/docs/content/user-manual/configuration/spring-boot-starter.cn.md
@@ -74,6 +74,7 @@ elasticjob.reg-center.namespace=elasticjob-springboot
| props | 否 |
| disabled | 否 |
| overwrite | 否 |
+| max-wait-millis | 否 |
**elasticJobClass 与 elasticJobType 互斥，每项作业只能有一种类型**
diff --git a/docs/content/user-manual/configuration/spring-boot-starter.en.md
b/docs/content/user-manual/configuration/spring-boot-starter.en.md
index 854c23b95..5b764aaab 100644
--- a/docs/content/user-manual/configuration/spring-boot-starter.en.md
+++ b/docs/content/user-manual/configuration/spring-boot-starter.en.md
@@ -74,6 +74,7 @@ Configuration:
| props | No |
| disabled | No |
| overwrite | No |
+| max-wait-millis | No |
**"elasticJobClass" and "elasticJobType" are mutually exclusive.**
diff --git a/docs/content/user-manual/configuration/spring-namespace.cn.md
b/docs/content/user-manual/configuration/spring-namespace.cn.md
index 874464502..901587c84 100644
--- a/docs/content/user-manual/configuration/spring-namespace.cn.md
+++ b/docs/content/user-manual/configuration/spring-namespace.cn.md
@@ -65,6 +65,7 @@ chapter = true
| props | 否 |
| disabled | 否 |
| overwrite | 否 |
+| max-wait-millis | 否 |
## 事件追踪配置
diff --git a/docs/content/user-manual/configuration/spring-namespace.en.md
b/docs/content/user-manual/configuration/spring-namespace.en.md
index 28e91b4dd..1d35a0736 100644
--- a/docs/content/user-manual/configuration/spring-namespace.en.md
+++ b/docs/content/user-manual/configuration/spring-namespace.en.md
@@ -64,6 +64,7 @@ Configuration:
| props | No |
| disabled | No |
| overwrite | No |
+| max-wait-millis | No |
## Event Tracing Configuration
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/BoundedWaitGuard.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/BoundedWaitGuard.java
new file mode 100644
index 000000000..0a94ef2a2
--- /dev/null
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/BoundedWaitGuard.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.infra.util;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Tracks elapsed time and interruption for a bounded wait loop, so callers
can consistently
+ * decide when to give up instead of blocking indefinitely.
+ */
+public final class BoundedWaitGuard {
+
+ private final long maxWaitNanos;
+
+ private final long startNanos;
+
+ private BoundedWaitGuard(final long maxWaitMillis) {
+ maxWaitNanos = TimeUnit.MILLISECONDS.toNanos(maxWaitMillis);
+ startNanos = System.nanoTime();
+ }
+
+ /**
+ * Start tracking a new bounded wait.
+ *
+ * @param maxWaitMillis max milliseconds to wait before giving up
+ * @return a new guard, timing starts now
+ */
+ public static BoundedWaitGuard start(final long maxWaitMillis) {
+ return new BoundedWaitGuard(maxWaitMillis);
+ }
+
+ /**
+ * Judge whether the caller should give up waiting: either the thread was
interrupted,
+ * or the configured max wait time has elapsed.
+ *
+ * @return whether to give up
+ */
+ public boolean shouldGiveUp() {
+ return Thread.currentThread().isInterrupted() || System.nanoTime() -
startNanos > maxWaitNanos;
+ }
+
+ /**
+ * Judge whether giving up is specifically due to interruption, so callers
can log accordingly.
+ *
+ * @return whether the current thread is interrupted
+ */
+ public boolean isInterrupted() {
+ return Thread.currentThread().isInterrupted();
+ }
+}
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/WaitTimeoutNotifier.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/WaitTimeoutNotifier.java
new file mode 100644
index 000000000..134353804
--- /dev/null
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/infra/util/WaitTimeoutNotifier.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.infra.util;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.spi.executor.error.handler.JobErrorHandler;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+
+/**
+ * Notifies the job's configured error handler when a bounded coordination
wait loop
+ * (sharding, leader election) gives up instead of blocking indefinitely.
+ */
+@Slf4j
+public final class WaitTimeoutNotifier {
+
+ private WaitTimeoutNotifier() {
+ }
+
+ /**
+ * Notify of a bounded-wait timeout.
+ *
+ * @param jobName job name
+ * @param jobConfig job configuration
+ * @param waitDescription description of what was being waited for, for
logging
+ */
+ public static void notifyTimeout(final String jobName, final
JobConfiguration jobConfig, final String waitDescription) {
+ String message = String.format("Job '%s' waited more than %d ms for
%s; giving up to avoid indefinite blocking.",
+ jobName, jobConfig.getMaxWaitMillis(), waitDescription);
+ log.error(message);
+ if (null == jobConfig.getJobErrorHandlerType()) {
+ return;
+ }
+ JobErrorHandler jobErrorHandler =
TypedSPILoader.getService(JobErrorHandler.class,
jobConfig.getJobErrorHandlerType(), jobConfig.getProps());
+ try {
+ jobErrorHandler.handleException(jobName, new
IllegalStateException(message));
+ } finally {
+ jobErrorHandler.close();
+ }
+ }
+}
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilder.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilder.java
index ac284830a..c213ddea0 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilder.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilder.java
@@ -59,7 +59,8 @@ public final class JobAnnotationBuilder {
.jobListenerTypes(annotation.jobListenerTypes())
.description(annotation.description())
.disabled(annotation.disabled())
- .overwrite(annotation.overwrite());
+ .overwrite(annotation.overwrite())
+ .maxWaitMillis(annotation.maxWaitMillis());
for (Class<? extends JobExtraConfigurationFactory> clazz :
annotation.extraConfigurations()) {
try {
Optional<JobExtraConfiguration> jobExtraConfig =
clazz.newInstance().getJobExtraConfiguration();
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJO.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJO.java
index 9d5ca39c9..01465b54b 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJO.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJO.java
@@ -81,6 +81,8 @@ public final class JobConfigurationPOJO {
private boolean staticSharding;
+ private long maxWaitMillis = 60_000L;
+
/**
* Convert to job configuration.
*
@@ -93,7 +95,7 @@ public final class JobConfigurationPOJO {
.maxTimeDiffSeconds(maxTimeDiffSeconds).reconcileIntervalMinutes(reconcileIntervalMinutes)
.jobShardingStrategyType(jobShardingStrategyType).jobExecutorThreadPoolSizeProviderType(jobExecutorThreadPoolSizeProviderType)
.jobErrorHandlerType(jobErrorHandlerType).jobListenerTypes(jobListenerTypes.toArray(new
String[]{})).description(description)
-
.disabled(disabled).overwrite(overwrite).label(label).staticSharding(staticSharding).build();
+
.disabled(disabled).overwrite(overwrite).label(label).staticSharding(staticSharding).maxWaitMillis(maxWaitMillis).build();
jobExtraConfigurations.stream().map(YamlConfiguration::toConfiguration).forEach(result.getExtraConfigurations()::add);
for (Object each : props.keySet()) {
result.getProps().setProperty(each.toString(),
props.get(each.toString()).toString());
@@ -133,6 +135,7 @@ public final class JobConfigurationPOJO {
result.setOverwrite(jobConfig.isOverwrite());
result.setLabel(jobConfig.getLabel());
result.setStaticSharding(jobConfig.isStaticSharding());
+ result.setMaxWaitMillis(jobConfig.getMaxWaitMillis());
return result;
}
}
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderService.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderService.java
index 5ffa7c2ea..1203b0d97 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderService.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderService.java
@@ -19,6 +19,11 @@ package
org.apache.shardingsphere.elasticjob.kernel.internal.election;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.exception.JobConfigurationException;
+import org.apache.shardingsphere.elasticjob.kernel.infra.util.BoundedWaitGuard;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.util.WaitTimeoutNotifier;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
import
org.apache.shardingsphere.elasticjob.kernel.internal.schedule.JobRegistry;
import
org.apache.shardingsphere.elasticjob.kernel.internal.sharding.JobInstance;
import
org.apache.shardingsphere.elasticjob.kernel.internal.server.ServerService;
@@ -39,10 +44,13 @@ public final class LeaderService {
private final JobNodeStorage jobNodeStorage;
+ private final ConfigurationService configService;
+
public LeaderService(final CoordinatorRegistryCenter regCenter, final
String jobName) {
this.jobName = jobName;
jobNodeStorage = new JobNodeStorage(regCenter, jobName);
serverService = new ServerService(regCenter, jobName);
+ configService = new ConfigurationService(regCenter, jobName);
}
/**
@@ -56,15 +64,29 @@ public final class LeaderService {
/**
* Judge current server is leader or not.
- *
+ *
* <p>
* If leader is electing, this method will block until leader elected
success.
* </p>
- *
+ *
* @return current server is leader or not
*/
public boolean isLeaderUntilBlock() {
+ JobConfiguration jobConfig = null;
+ BoundedWaitGuard waitGuard = null;
while (!hasLeader() && serverService.hasAvailableServers()) {
+ if (null == waitGuard) {
+ try {
+ jobConfig = configService.load(true);
+ } catch (final JobConfigurationException ex) {
+ return isLeader();
+ }
+ waitGuard =
BoundedWaitGuard.start(jobConfig.getMaxWaitMillis());
+ }
+ if (waitGuard.shouldGiveUp()) {
+ WaitTimeoutNotifier.notifyTimeout(jobName, jobConfig,
waitGuard.isInterrupted() ? "leader election to complete (interrupted)" :
"leader election to complete");
+ return isLeader();
+ }
log.info("Leader is electing, waiting for {} ms", 100);
BlockUtils.waitingShortTime();
if (!JobRegistry.getInstance().isShutdown(jobName) &&
serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getServerIp()))
{
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingService.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingService.java
index 61f596e2a..d83953abf 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingService.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingService.java
@@ -19,7 +19,10 @@ package
org.apache.shardingsphere.elasticjob.kernel.internal.sharding;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.exception.JobConfigurationException;
import org.apache.shardingsphere.elasticjob.kernel.infra.util.BlockUtils;
+import org.apache.shardingsphere.elasticjob.kernel.infra.util.BoundedWaitGuard;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.util.WaitTimeoutNotifier;
import
org.apache.shardingsphere.elasticjob.kernel.internal.sharding.strategy.JobShardingStrategy;
import org.apache.shardingsphere.elasticjob.kernel.infra.yaml.YamlEngine;
import
org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
@@ -121,19 +124,52 @@ public final class ShardingService {
jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, "");
resetShardingInfo(shardingTotalCount);
JobShardingStrategy jobShardingStrategy =
TypedSPILoader.getService(JobShardingStrategy.class,
jobConfig.getJobShardingStrategyType());
-
jobNodeStorage.executeInTransaction(getShardingResultTransactionOperations(jobShardingStrategy.sharding(availableJobInstances,
jobName, shardingTotalCount)));
+ boolean success =
jobNodeStorage.executeInTransactionStrict(getShardingResultTransactionOperations(jobShardingStrategy.sharding(availableJobInstances,
jobName, shardingTotalCount)));
+ if (!success) {
+ log.error("Job '{}' sharding transaction failed due to a registry
center exception; clearing PROCESSING flag to avoid indefinite blocking of
other instances.", jobName);
+ jobNodeStorage.removeJobNodeIfExisted(ShardingNode.PROCESSING);
+ return;
+ }
log.debug("Job '{}' sharding complete.", jobName);
}
private void blockUntilShardingCompleted() {
+ JobConfiguration jobConfig = null;
+ BoundedWaitGuard waitGuard = null;
while (!leaderService.isLeaderUntilBlock() &&
(jobNodeStorage.isJobNodeExisted(ShardingNode.NECESSARY) ||
jobNodeStorage.isJobNodeExisted(ShardingNode.PROCESSING))) {
+ if (null == waitGuard) {
+ try {
+ jobConfig = configService.load(true);
+ } catch (final JobConfigurationException ex) {
+ return;
+ }
+ waitGuard =
BoundedWaitGuard.start(jobConfig.getMaxWaitMillis());
+ }
+ if (waitGuard.shouldGiveUp()) {
+ WaitTimeoutNotifier.notifyTimeout(jobName, jobConfig,
waitGuard.isInterrupted() ? "sharding to complete (interrupted)" : "sharding to
complete");
+ return;
+ }
log.debug("Job '{}' sleep short time until sharding completed.",
jobName);
BlockUtils.waitingShortTime();
}
}
private void waitingOtherShardingItemCompleted() {
+ JobConfiguration jobConfig = null;
+ BoundedWaitGuard waitGuard = null;
while (executionService.hasRunningItems()) {
+ if (null == waitGuard) {
+ try {
+ jobConfig = configService.load(true);
+ } catch (final JobConfigurationException ex) {
+ return;
+ }
+ waitGuard =
BoundedWaitGuard.start(jobConfig.getMaxWaitMillis());
+ }
+ if (waitGuard.shouldGiveUp()) {
+ WaitTimeoutNotifier.notifyTimeout(jobName, jobConfig,
waitGuard.isInterrupted() ? "other job items to complete (interrupted)" :
"other job items to complete");
+ return;
+ }
log.debug("Job '{}' sleep short time until other job completed.",
jobName);
BlockUtils.waitingShortTime();
}
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/storage/JobNodeStorage.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/storage/JobNodeStorage.java
index 6ff43fe63..0c6468612 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/storage/JobNodeStorage.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/storage/JobNodeStorage.java
@@ -195,6 +195,31 @@ public final class JobNodeStorage {
}
}
+ /**
+ * Execute operations in transaction, reporting failure instead of
swallowing it silently.
+ *
+ * <p>
+ * Only exceptions that the registry center treats as ignorable are
reported as {@code false}. Any other exception is still rethrown, as in {@link
#executeInTransaction(List)}.
+ * </p>
+ *
+ * @param transactionOperations operations to be executed in transaction
+ * @return true if transaction succeeded, false if it failed with an
ignored registry-center exception
+ */
+ public boolean executeInTransactionStrict(final List<TransactionOperation>
transactionOperations) {
+ List<TransactionOperation> result = new
ArrayList<>(transactionOperations.size() + 1);
+ result.add(TransactionOperation.opCheckExists("/"));
+ result.addAll(transactionOperations);
+ try {
+ regCenter.executeInTransaction(result);
+ return true;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ return false;
+ }
+ }
+
/**
* Execute in leader server.
*
diff --git
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListener.java
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListener.java
index 64b0c6e01..98718c41b 100644
---
a/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListener.java
+++
b/kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListener.java
@@ -69,7 +69,7 @@ public abstract class
AbstractDistributeOnceElasticJobListener implements Elasti
startedWait.wait(startedTimeoutMilliseconds);
}
} catch (final InterruptedException ex) {
- Thread.interrupted();
+ Thread.currentThread().interrupt();
}
if (timeService.getCurrentMillis() - before >=
startedTimeoutMilliseconds) {
guaranteeService.clearAllStartedInfo();
@@ -95,7 +95,7 @@ public abstract class
AbstractDistributeOnceElasticJobListener implements Elasti
completedWait.wait(completedTimeoutMilliseconds);
}
} catch (final InterruptedException ex) {
- Thread.interrupted();
+ Thread.currentThread().interrupt();
}
if (timeService.getCurrentMillis() - before >=
completedTimeoutMilliseconds) {
guaranteeService.clearAllCompletedInfo();
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilderMaxWaitMillisTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilderMaxWaitMillisTest.java
new file mode 100644
index 000000000..62220e011
--- /dev/null
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/annotation/JobAnnotationBuilderMaxWaitMillisTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.internal.annotation;
+
+import org.apache.shardingsphere.elasticjob.annotation.ElasticJobConfiguration;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JobAnnotationBuilderMaxWaitMillisTest {
+
+ @Test
+ void assertGenerateJobConfigurationWithDefaultMaxWaitMillis() {
+ assertEquals(60_000L,
JobAnnotationBuilder.generateJobConfiguration(DefaultJob.class).getMaxWaitMillis());
+ }
+
+ @Test
+ void assertGenerateJobConfigurationWithCustomMaxWaitMillis() {
+ assertEquals(45_000L,
JobAnnotationBuilder.generateJobConfiguration(CustomJob.class).getMaxWaitMillis());
+ }
+
+ @ElasticJobConfiguration(jobName = "default_wait_job", shardingTotalCount
= 1)
+ private static final class DefaultJob {
+ }
+
+ @ElasticJobConfiguration(jobName = "custom_wait_job", shardingTotalCount =
1, maxWaitMillis = 45_000L)
+ private static final class CustomJob {
+ }
+}
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOMaxWaitMillisTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOMaxWaitMillisTest.java
new file mode 100644
index 000000000..72d731720
--- /dev/null
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOMaxWaitMillisTest.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.internal.config;
+
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JobConfigurationPOJOMaxWaitMillisTest {
+
+ @Test
+ void assertMaxWaitMillisSurvivesRoundTrip() {
+ JobConfiguration original = JobConfiguration.newBuilder("test_job",
1).maxWaitMillis(12345L).build();
+
+ JobConfigurationPOJO pojo =
JobConfigurationPOJO.fromJobConfiguration(original);
+ assertEquals(12345L, pojo.getMaxWaitMillis(), "POJO should preserve
maxWaitMillis when converting from JobConfiguration");
+
+ JobConfiguration roundTripped = pojo.toJobConfiguration();
+ assertEquals(12345L, roundTripped.getMaxWaitMillis(), "maxWaitMillis
should survive a full round trip through JobConfigurationPOJO");
+ }
+}
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOTest.java
index b7605b382..f1b8c6d2a 100644
---
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOTest.java
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/config/JobConfigurationPOJOTest.java
@@ -42,6 +42,7 @@ class JobConfigurationPOJOTest {
+ "jobParameter: param\n"
+ "jobShardingStrategyType: AVG_ALLOCATION\n"
+ "maxTimeDiffSeconds: -1\n"
+ + "maxWaitMillis: 60000\n"
+ "misfire: false\n"
+ "monitorExecution: false\n"
+ "overwrite: false\n"
@@ -57,6 +58,7 @@ class JobConfigurationPOJOTest {
+ "failover: false\n"
+ "jobName: test_job\n"
+ "maxTimeDiffSeconds: -1\n"
+ + "maxWaitMillis: 60000\n"
+ "misfire: false\n"
+ "monitorExecution: false\n"
+ "overwrite: false\n"
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceBoundedWaitTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceBoundedWaitTest.java
new file mode 100644
index 000000000..979ea5f45
--- /dev/null
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceBoundedWaitTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.internal.election;
+
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.exception.JobConfigurationException;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.server.ServerService;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class LeaderServiceBoundedWaitTest {
+
+ @Test
+ void assertIsLeaderUntilBlockExitsWithinBoundedTime() throws
ReflectiveOperationException, InterruptedException {
+ LeaderService leaderService = createLeaderServiceWithNoLeader(500L);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startWorker(leaderService, returned);
+ worker.join(3000L);
+ assertTrue(returned.get(), "isLeaderUntilBlock() should give up within
a bounded time instead of spinning forever");
+ }
+
+ @Test
+ void assertIsLeaderUntilBlockExitsPromptlyOnInterrupt() throws
ReflectiveOperationException, InterruptedException {
+ LeaderService leaderService = createLeaderServiceWithNoLeader(60_000L);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startWorker(leaderService, returned);
+ Thread.sleep(150L);
+ worker.interrupt();
+ worker.join(2000L);
+ assertTrue(returned.get(), "isLeaderUntilBlock() should exit promptly
when interrupted, well before the configured timeout");
+ }
+
+ @Test
+ void assertIsLeaderUntilBlockExitsWhenJobConfigurationIsRemoved() throws
ReflectiveOperationException, InterruptedException {
+ LeaderService leaderService = createLeaderServiceWithNoLeader(60_000L);
+ ConfigurationService configService = mock(ConfigurationService.class);
+ when(configService.load(anyBoolean())).thenThrow(new
JobConfigurationException("removed"));
+ setPrivateField(leaderService, "configService", configService);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startWorker(leaderService, returned);
+ worker.join(2000L);
+ assertTrue(returned.get(), "isLeaderUntilBlock() should stop waiting
without throwing when the job configuration has been removed");
+ }
+
+ private LeaderService createLeaderServiceWithNoLeader(final long
maxWaitMillis) throws ReflectiveOperationException {
+ LeaderService result = new
LeaderService(mock(CoordinatorRegistryCenter.class), "bounded_wait_job");
+ ServerService serverService = mock(ServerService.class);
+ when(serverService.hasAvailableServers()).thenReturn(true);
+ setPrivateField(result, "serverService", serverService);
+ ConfigurationService configService = mock(ConfigurationService.class);
+
when(configService.load(anyBoolean())).thenReturn(JobConfiguration.newBuilder("bounded_wait_job",
1).maxWaitMillis(maxWaitMillis).build());
+ setPrivateField(result, "configService", configService);
+ return result;
+ }
+
+ private Thread startWorker(final LeaderService leaderService, final
AtomicBoolean returned) {
+ Thread result = new Thread(() -> {
+ leaderService.isLeaderUntilBlock();
+ returned.set(true);
+ });
+ result.setDaemon(true);
+ result.start();
+ return result;
+ }
+
+ private static void setPrivateField(final Object target, final String
fieldName, final Object value) throws ReflectiveOperationException {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+}
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceTest.java
index b1fbe3fa5..96a82c015 100644
---
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceTest.java
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/election/LeaderServiceTest.java
@@ -17,6 +17,8 @@
package org.apache.shardingsphere.elasticjob.kernel.internal.election;
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
import
org.apache.shardingsphere.elasticjob.kernel.internal.sharding.JobInstance;
import
org.apache.shardingsphere.elasticjob.kernel.internal.election.LeaderService.LeaderElectionExecutionCallback;
import
org.apache.shardingsphere.elasticjob.kernel.internal.schedule.JobRegistry;
@@ -54,6 +56,9 @@ class LeaderServiceTest {
@Mock
private ServerService serverService;
+ @Mock
+ private ConfigurationService configService;
+
private LeaderService leaderService;
@BeforeEach
@@ -62,6 +67,7 @@ class LeaderServiceTest {
leaderService = new LeaderService(null, "test_job");
ReflectionUtils.setFieldValue(leaderService, "jobNodeStorage",
jobNodeStorage);
ReflectionUtils.setFieldValue(leaderService, "serverService",
serverService);
+ ReflectionUtils.setFieldValue(leaderService, "configService",
configService);
}
@Test
@@ -102,6 +108,7 @@ class LeaderServiceTest {
when(jobNodeStorage.isJobNodeExisted("leader/election/instance")).thenReturn(false,
true);
when(serverService.hasAvailableServers()).thenReturn(true);
when(serverService.isAvailableServer("127.0.0.1")).thenReturn(true);
+
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job",
1).build());
when(jobNodeStorage.getJobNodeData("leader/election/instance")).thenReturn("127.0.0.1@-@0");
assertTrue(leaderService.isLeaderUntilBlock());
verify(jobNodeStorage).executeInLeader(eq("leader/election/latch"),
ArgumentMatchers.<LeaderElectionExecutionCallback>any());
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceBoundedWaitTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceBoundedWaitTest.java
new file mode 100644
index 000000000..80c16d335
--- /dev/null
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceBoundedWaitTest.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.internal.sharding;
+
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import
org.apache.shardingsphere.elasticjob.kernel.infra.exception.JobConfigurationException;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.config.ConfigurationService;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.election.LeaderService;
+import
org.apache.shardingsphere.elasticjob.kernel.internal.storage.JobNodeStorage;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ShardingServiceBoundedWaitTest {
+
+ @Test
+ void assertBlockUntilShardingCompletedExitsWithinBoundedTime() throws
ReflectiveOperationException, InterruptedException {
+ ShardingService shardingService =
createShardingServiceWithStuckProcessingNode(500L);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startBlockingWorker(shardingService, returned);
+ worker.join(3000L);
+ assertTrue(returned.get(), "blockUntilShardingCompleted() should give
up within a bounded time instead of spinning forever");
+ }
+
+ @Test
+ void assertBlockUntilShardingCompletedExitsPromptlyOnInterrupt() throws
ReflectiveOperationException, InterruptedException {
+ ShardingService shardingService =
createShardingServiceWithStuckProcessingNode(60_000L);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startBlockingWorker(shardingService, returned);
+ Thread.sleep(150L);
+ worker.interrupt();
+ worker.join(2000L);
+ assertTrue(returned.get(), "blockUntilShardingCompleted() should exit
promptly when interrupted, well before the configured timeout");
+ }
+
+ @Test
+ void assertBlockUntilShardingCompletedExitsWhenJobConfigurationIsRemoved()
throws ReflectiveOperationException, InterruptedException {
+ ShardingService shardingService =
createShardingServiceWithStuckProcessingNode(60_000L);
+ ConfigurationService configService = mock(ConfigurationService.class);
+ when(configService.load(anyBoolean())).thenThrow(new
JobConfigurationException("removed"));
+ setPrivateField(shardingService, "configService", configService);
+ AtomicBoolean returned = new AtomicBoolean(false);
+ Thread worker = startBlockingWorker(shardingService, returned);
+ worker.join(2000L);
+ assertTrue(returned.get(), "blockUntilShardingCompleted() should stop
waiting without throwing when the job configuration has been removed");
+ }
+
+ private ShardingService createShardingServiceWithStuckProcessingNode(final
long maxWaitMillis) throws ReflectiveOperationException {
+ ShardingService result = new
ShardingService(mock(CoordinatorRegistryCenter.class), "bounded_wait_job");
+ LeaderService leaderService = mock(LeaderService.class);
+ when(leaderService.isLeaderUntilBlock()).thenReturn(false);
+ setPrivateField(result, "leaderService", leaderService);
+ JobNodeStorage jobNodeStorage = mock(JobNodeStorage.class);
+
when(jobNodeStorage.isJobNodeExisted(ShardingNode.NECESSARY)).thenReturn(true);
+
when(jobNodeStorage.isJobNodeExisted(ShardingNode.PROCESSING)).thenReturn(true);
+ setPrivateField(result, "jobNodeStorage", jobNodeStorage);
+ ConfigurationService configService = mock(ConfigurationService.class);
+
when(configService.load(anyBoolean())).thenReturn(JobConfiguration.newBuilder("bounded_wait_job",
1).maxWaitMillis(maxWaitMillis).build());
+ setPrivateField(result, "configService", configService);
+ return result;
+ }
+
+ private Thread startBlockingWorker(final ShardingService shardingService,
final AtomicBoolean returned) throws ReflectiveOperationException {
+ Method method =
ShardingService.class.getDeclaredMethod("blockUntilShardingCompleted");
+ method.setAccessible(true);
+ Thread result = new Thread(() -> {
+ try {
+ method.invoke(shardingService);
+ returned.set(true);
+ } catch (final IllegalAccessException | InvocationTargetException
ex) {
+ throw new IllegalStateException(ex);
+ }
+ });
+ result.setDaemon(true);
+ result.start();
+ return result;
+ }
+
+ private static void setPrivateField(final Object target, final String
fieldName, final Object value) throws ReflectiveOperationException {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+}
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceTest.java
index f0d9c9eaa..aa478fdf1 100644
---
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceTest.java
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ShardingServiceTest.java
@@ -127,6 +127,7 @@ class ShardingServiceTest {
when(instanceService.getAvailableJobInstances()).thenReturn(Collections.singletonList(new
JobInstance("127.0.0.1@-@0")));
when(leaderService.isLeaderUntilBlock()).thenReturn(false);
when(jobNodeStorage.isJobNodeExisted("leader/sharding/processing")).thenReturn(true,
false);
+
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job",
3).cron("0/1 * * * * ?").build());
shardingService.shardingIfNecessary();
verify(jobNodeStorage,
times(0)).fillEphemeralJobNode(ShardingNode.PROCESSING, "");
}
@@ -136,9 +137,11 @@ class ShardingServiceTest {
when(instanceService.getAvailableJobInstances()).thenReturn(Collections.singletonList(new
JobInstance("127.0.0.1@-@0")));
when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true);
when(leaderService.isLeaderUntilBlock()).thenReturn(true);
+
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job",
3).cron("0/1 * * * * ?").monitorExecution(true).build());
when(configService.load(false)).thenReturn(JobConfiguration.newBuilder("test_job",
3).cron("0/1 * * * * ?").monitorExecution(true).build());
when(executionService.hasRunningItems()).thenReturn(true, false);
when(jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT)).thenReturn(Arrays.asList("0",
"1"));
+
when(jobNodeStorage.executeInTransactionStrict(any(List.class))).thenReturn(true);
shardingService.shardingIfNecessary();
verify(executionService, times(2)).hasRunningItems();
verify(jobNodeStorage).removeJobNodeIfExisted("sharding/0/instance");
@@ -148,7 +151,7 @@ class ShardingServiceTest {
verify(jobNodeStorage).removeJobNodeIfExisted("sharding/2/instance");
verify(jobNodeStorage).createJobNodeIfNeeded("sharding/2");
verify(jobNodeStorage).fillEphemeralJobNode("leader/sharding/processing", "");
- verify(jobNodeStorage).executeInTransaction(any(List.class));
+ verify(jobNodeStorage).executeInTransactionStrict(any(List.class));
}
@Test
@@ -158,6 +161,7 @@ class ShardingServiceTest {
when(leaderService.isLeaderUntilBlock()).thenReturn(true);
when(configService.load(false)).thenReturn(JobConfiguration.newBuilder("test_job",
3).cron("0/1 * * * * ?").monitorExecution(false).build());
when(jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT)).thenReturn(Arrays.asList("0",
"1", "2", "3"));
+
when(jobNodeStorage.executeInTransactionStrict(any(List.class))).thenReturn(true);
shardingService.shardingIfNecessary();
verify(jobNodeStorage).removeJobNodeIfExisted("sharding/0/instance");
verify(jobNodeStorage).createJobNodeIfNeeded("sharding/0");
@@ -168,7 +172,19 @@ class ShardingServiceTest {
verify(jobNodeStorage, times(0)).removeJobNodeIfExisted("execution/2");
verify(jobNodeStorage).removeJobNodeIfExisted("sharding/3");
verify(jobNodeStorage).fillEphemeralJobNode("leader/sharding/processing", "");
- verify(jobNodeStorage).executeInTransaction(any(List.class));
+ verify(jobNodeStorage).executeInTransactionStrict(any(List.class));
+ }
+
+ @Test
+ void assertShardingClearsProcessingNodeWhenTransactionFails() {
+
when(instanceService.getAvailableJobInstances()).thenReturn(Collections.singletonList(new
JobInstance("127.0.0.1@-@0")));
+
when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true);
+ when(leaderService.isLeaderUntilBlock()).thenReturn(true);
+
when(configService.load(false)).thenReturn(JobConfiguration.newBuilder("test_job",
3).cron("0/1 * * * * ?").monitorExecution(false).build());
+
when(jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT)).thenReturn(Arrays.asList("0",
"1", "2"));
+
when(jobNodeStorage.executeInTransactionStrict(any(List.class))).thenReturn(false);
+ shardingService.shardingIfNecessary();
+ verify(jobNodeStorage).removeJobNodeIfExisted(ShardingNode.PROCESSING);
}
@Test
diff --git
a/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListenerBoundedWaitTest.java
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListenerBoundedWaitTest.java
new file mode 100644
index 000000000..ff2f3f2ae
--- /dev/null
+++
b/kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/listener/AbstractDistributeOnceElasticJobListenerBoundedWaitTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.kernel.listener;
+
+import
org.apache.shardingsphere.elasticjob.kernel.internal.guarantee.GuaranteeService;
+import
org.apache.shardingsphere.elasticjob.spi.listener.param.ShardingContexts;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class AbstractDistributeOnceElasticJobListenerBoundedWaitTest {
+
+ @Test
+ void assertBeforeJobExecutedRestoresInterruptStatus() {
+ TestListener listener = new TestListener();
+ GuaranteeService guaranteeService = mock(GuaranteeService.class);
+ when(guaranteeService.isRegisterStartSuccess(any())).thenReturn(true);
+ when(guaranteeService.isAllStarted()).thenReturn(false);
+ listener.setGuaranteeService(guaranteeService);
+ ShardingContexts shardingContexts = createShardingContexts();
+ Thread.currentThread().interrupt();
+ try {
+ listener.beforeJobExecuted(shardingContexts);
+ assertTrue(Thread.currentThread().isInterrupted(), "interrupt
status should be restored after an interrupted wait");
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ @Test
+ void assertAfterJobExecutedRestoresInterruptStatus() {
+ TestListener listener = new TestListener();
+ GuaranteeService guaranteeService = mock(GuaranteeService.class);
+
when(guaranteeService.isRegisterCompleteSuccess(any())).thenReturn(true);
+ when(guaranteeService.isAllCompleted()).thenReturn(false);
+ listener.setGuaranteeService(guaranteeService);
+ ShardingContexts shardingContexts = createShardingContexts();
+ Thread.currentThread().interrupt();
+ try {
+ listener.afterJobExecuted(shardingContexts);
+ assertTrue(Thread.currentThread().isInterrupted(), "interrupt
status should be restored after an interrupted wait");
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ private ShardingContexts createShardingContexts() {
+ ShardingContexts result = mock(ShardingContexts.class);
+
when(result.getShardingItemParameters()).thenReturn(Collections.singletonMap(0,
"0"));
+ return result;
+ }
+
+ private static final class TestListener extends
AbstractDistributeOnceElasticJobListener {
+
+ TestListener() {
+ super(60_000L, 60_000L);
+ }
+
+ @Override
+ public void doBeforeJobExecutedAtLastStarted(final ShardingContexts
shardingContexts) {
+ }
+
+ @Override
+ public void doAfterJobExecutedAtLastCompleted(final ShardingContexts
shardingContexts) {
+ }
+
+ @Override
+ public String getType() {
+ return "TestListener";
+ }
+ }
+}
diff --git
a/lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lifecycle/fixture/LifecycleYamlConstants.java
b/lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lifecycle/fixture/LifecycleYamlConstants.java
index bf4b710da..23ff6470d 100644
---
a/lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lifecycle/fixture/LifecycleYamlConstants.java
+++
b/lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lifecycle/fixture/LifecycleYamlConstants.java
@@ -43,6 +43,7 @@ public final class LifecycleYamlConstants {
+ "jobName: test_job\n"
+ "jobParameter: param\n"
+ "maxTimeDiffSeconds: -1\n"
+ + "maxWaitMillis: 60000\n"
+ "misfire: true\n"
+ "monitorExecution: true\n"
+ "overwrite: false\n"
diff --git
a/spring/boot-starter/src/main/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationProperties.java
b/spring/boot-starter/src/main/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationProperties.java
index b1ab7691f..76db14644 100644
---
a/spring/boot-starter/src/main/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationProperties.java
+++
b/spring/boot-starter/src/main/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationProperties.java
@@ -75,6 +75,8 @@ public final class ElasticJobConfigurationProperties {
private boolean overwrite;
+ private long maxWaitMillis = 60_000L;
+
/**
* Convert to job configuration.
*
@@ -87,7 +89,7 @@ public final class ElasticJobConfigurationProperties {
.monitorExecution(monitorExecution).failover(failover).misfire(misfire)
.maxTimeDiffSeconds(maxTimeDiffSeconds).reconcileIntervalMinutes(reconcileIntervalMinutes)
.jobShardingStrategyType(jobShardingStrategyType).jobExecutorThreadPoolSizeProviderType(jobExecutorThreadPoolSizeProviderType).jobErrorHandlerType(jobErrorHandlerType)
- .jobListenerTypes(jobListenerTypes.toArray(new
String[0])).description(description).disabled(disabled).overwrite(overwrite).build();
+ .jobListenerTypes(jobListenerTypes.toArray(new
String[0])).description(description).disabled(disabled).overwrite(overwrite).maxWaitMillis(maxWaitMillis).build();
props.stringPropertyNames().forEach(each ->
result.getProps().setProperty(each, props.getProperty(each)));
return result;
}
diff --git
a/spring/boot-starter/src/test/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationPropertiesMaxWaitMillisTest.java
b/spring/boot-starter/src/test/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationPropertiesMaxWaitMillisTest.java
new file mode 100644
index 000000000..34c4c83bc
--- /dev/null
+++
b/spring/boot-starter/src/test/java/org/apache/shardingsphere/elasticjob/spring/boot/job/ElasticJobConfigurationPropertiesMaxWaitMillisTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.spring.boot.job;
+
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ElasticJobConfigurationPropertiesMaxWaitMillisTest {
+
+ @Test
+ void assertMaxWaitMillisPropagatesToJobConfiguration() {
+ ElasticJobConfigurationProperties properties = new
ElasticJobConfigurationProperties();
+ properties.setShardingTotalCount(3);
+ properties.setMaxWaitMillis(45000L);
+
+ JobConfiguration actual = properties.toJobConfiguration("test_job");
+
+ assertEquals(45000L, actual.getMaxWaitMillis(), "maxWaitMillis set via
Spring Boot properties should propagate to the built JobConfiguration");
+ }
+
+ @Test
+ void assertMaxWaitMillisDefaultsWhenNotSet() {
+ ElasticJobConfigurationProperties properties = new
ElasticJobConfigurationProperties();
+ properties.setShardingTotalCount(3);
+
+ JobConfiguration actual = properties.toJobConfiguration("test_job");
+
+ assertEquals(60_000L, actual.getMaxWaitMillis(), "maxWaitMillis should
default to 60000ms when not explicitly configured via Spring Boot properties");
+ }
+}
diff --git
a/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParser.java
b/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParser.java
index 38711504e..5d2d1f24a 100644
---
a/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParser.java
+++
b/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParser.java
@@ -96,6 +96,7 @@ public final class JobBeanDefinitionParser extends
AbstractBeanDefinitionParser
result.addConstructorArgValue(element.getAttribute(JobBeanDefinitionTag.OVERWRITE_ATTRIBUTE));
result.addConstructorArgValue(element.getAttribute(JobBeanDefinitionTag.LABEL_ATTRIBUTE));
result.addConstructorArgValue(element.getAttribute(JobBeanDefinitionTag.STATIC_SHARDING_ATTRIBUTE));
+
result.addConstructorArgValue(element.getAttribute(JobBeanDefinitionTag.MAX_WAIT_MILLIS_ATTRIBUTE));
return result.getBeanDefinition();
}
diff --git
a/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/tag/JobBeanDefinitionTag.java
b/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/tag/JobBeanDefinitionTag.java
index a74c2d311..a0435e591 100644
---
a/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/tag/JobBeanDefinitionTag.java
+++
b/spring/namespace/src/main/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/tag/JobBeanDefinitionTag.java
@@ -73,4 +73,6 @@ public final class JobBeanDefinitionTag {
public static final String LABEL_ATTRIBUTE = "label";
public static final String STATIC_SHARDING_ATTRIBUTE = "static-sharding";
+
+ public static final String MAX_WAIT_MILLIS_ATTRIBUTE = "max-wait-millis";
}
diff --git
a/spring/namespace/src/main/resources/META-INF/namespace/elasticjob.xsd
b/spring/namespace/src/main/resources/META-INF/namespace/elasticjob.xsd
index 308a8e460..86f893c6a 100644
--- a/spring/namespace/src/main/resources/META-INF/namespace/elasticjob.xsd
+++ b/spring/namespace/src/main/resources/META-INF/namespace/elasticjob.xsd
@@ -64,6 +64,7 @@
<xsd:attribute name="overwrite" type="xsd:string"
default="false" />
<xsd:attribute name="label" type="xsd:string" />
<xsd:attribute name="static-sharding" type="xsd:string"
default="false" />
+ <xsd:attribute name="max-wait-millis" type="xsd:string"
default="60000" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
diff --git
a/spring/namespace/src/test/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParserMaxWaitMillisTest.java
b/spring/namespace/src/test/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParserMaxWaitMillisTest.java
new file mode 100644
index 000000000..b86df2a05
--- /dev/null
+++
b/spring/namespace/src/test/java/org/apache/shardingsphere/elasticjob/spring/namespace/job/parser/JobBeanDefinitionParserMaxWaitMillisTest.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.elasticjob.spring.namespace.job.parser;
+
+import
org.apache.shardingsphere.elasticjob.spring.namespace.job.tag.JobBeanDefinitionTag;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JobBeanDefinitionParserMaxWaitMillisTest {
+
+ @Test
+ void assertMaxWaitMillisAttributeName() {
+ assertEquals("max-wait-millis",
JobBeanDefinitionTag.MAX_WAIT_MILLIS_ATTRIBUTE);
+ }
+}
