You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by cs...@apache.org on 2026/08/26 06:42:48 UTC
(celeborn) branch CELEBORN-2445 created (now 6462d76bd)
This is an automated email from the ASF dual-hosted git repository.
cxzl25 pushed a change to branch CELEBORN-2445
in repository https://gitbox.apache.org/repos/asf/celeborn.git
at 6462d76bd feat(worker): add per-decommission timeout override
This branch includes the following new commits:
new 6462d76bd feat(worker): add per-decommission timeout override
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(celeborn) 01/01: feat(worker): add per-decommission timeout override
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cxzl25 pushed a commit to branch CELEBORN-2445
in repository https://gitbox.apache.org/repos/asf/celeborn.git
commit 6462d76bd29dcd1d5971684a5fa4f9480a1cff97
Author: sychen <sy...@ctrip.com>
AuthorDate: Wed Aug 26 14:42:37 2026 +0800
feat(worker): add per-decommission timeout override
---
.../apache/celeborn/cli/worker/WorkerOptions.scala | 8 +++++
.../celeborn/cli/worker/WorkerSubcommandImpl.scala | 11 +++++--
.../org/apache/celeborn/common/CelebornConf.scala | 6 +++-
docs/celeborn_cli.md | 3 ++
docs/decommissioning.md | 11 +++++++
docs/restapi.md | 2 +-
.../celeborn/rest/v1/model/WorkerExitRequest.java | 38 ++++++++++++++++++++--
.../src/main/openapi3/worker_rest_v1.yaml | 5 +++
.../celeborn/server/common/HttpService.scala | 4 ++-
.../celeborn/service/deploy/worker/Worker.scala | 38 +++++++++++++++++++---
.../deploy/worker/http/api/ApiWorkerResource.scala | 10 ++++--
.../deploy/worker/http/api/v1/WorkerResource.scala | 6 ++--
12 files changed, 124 insertions(+), 18 deletions(-)
diff --git a/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerOptions.scala b/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerOptions.scala
index 3d0ec7122..26f95bc74 100644
--- a/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerOptions.scala
+++ b/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerOptions.scala
@@ -61,6 +61,14 @@ final class WorkerOptions {
description = Array("Exit the application with a specified type"))
private[worker] var exitType: String = _
+ @Option(
+ names = Array("--exit-timeout"),
+ paramLabel = "duration",
+ description = Array(
+ "Optional decommission forceExitTimeout override for --exit DECOMMISSION " +
+ "(e.g. 600s/30m/1h); ignored for other exit types."))
+ private[worker] var exitTimeout: String = _
+
@Option(names = Array("--show-conf"), description = Array("Show worker conf"))
private[worker] var showConf: Boolean = _
diff --git a/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerSubcommandImpl.scala b/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerSubcommandImpl.scala
index f8f31615c..76a4b5ad9 100644
--- a/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerSubcommandImpl.scala
+++ b/cli/src/main/scala/org/apache/celeborn/cli/worker/WorkerSubcommandImpl.scala
@@ -68,8 +68,15 @@ class WorkerSubcommandImpl extends WorkerSubcommand {
private[worker] def runExit: HandleResponse = {
val workerExitType: TypeEnum = TypeEnum.valueOf(workerOptions.exitType)
- val workerExitRequest: WorkerExitRequest = new WorkerExitRequest().`type`(workerExitType)
- logInfo(s"Sending worker exit type: ${workerExitType.getValue}")
+ val workerExitRequest: WorkerExitRequest =
+ new WorkerExitRequest().`type`(workerExitType)
+ if (workerOptions.exitTimeout != null && workerOptions.exitTimeout.nonEmpty) {
+ workerExitRequest.timeout(workerOptions.exitTimeout)
+ logInfo(s"Sending worker exit type: ${workerExitType.getValue}, " +
+ s"timeout: ${workerOptions.exitTimeout}")
+ } else {
+ logInfo(s"Sending worker exit type: ${workerExitType.getValue}")
+ }
workerApi.workerExit(workerExitRequest, commonOptions.getAuthHeader)
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 2a928c611..81de1054f 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -4503,7 +4503,11 @@ object CelebornConf extends Logging {
val WORKER_DECOMMISSION_FORCE_EXIT_TIMEOUT: ConfigEntry[Long] =
buildConf("celeborn.worker.decommission.forceExitTimeout")
.categories("worker")
- .doc("The wait time of waiting for all the shuffle expire during worker decommission.")
+ .doc("The wait time of waiting for all the shuffle expire during worker decommission. " +
+ "This value is read once at decommission start and is not dynamically changeable for " +
+ "the running decommission; it can be overridden per-decommission via the optional " +
+ "`timeout` parameter of the worker exit API (or the CLI --exit-timeout option) without " +
+ "restarting the worker.")
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("6h")
diff --git a/docs/celeborn_cli.md b/docs/celeborn_cli.md
index 31b893f3a..6d1245731 100644
--- a/docs/celeborn_cli.md
+++ b/docs/celeborn_cli.md
@@ -232,6 +232,9 @@ Usage: celeborn-cli worker [-hV] [--apps=appId] [--auth-header=authHeader]
The comma separated dynamic configs to delete.
--delete-dynamic-conf Delete dynamic worker conf
--exit=exit_type Exit the application with a specified type
+ --exit-timeout=duration
+ Optional decommission forceExitTimeout override for --exit DECOMMISSION
+ (e.g. 600s/30m/1h); ignored for other exit types.
-h, --help Show this help message and exit.
--host-list=h1,h2,h3...
List of hosts to pass to the command
diff --git a/docs/decommissioning.md b/docs/decommissioning.md
index 2802034d8..d65f5bebf 100644
--- a/docs/decommissioning.md
+++ b/docs/decommissioning.md
@@ -54,6 +54,11 @@ to meet varying workload demands.
| celeborn.worker.decommission.forceExitTimeout | 6h |
| celeborn.worker.decommission.checkInterval | 30s |
+> `forceExitTimeout` is read once at decommission start and is **not dynamically changeable**
+> for the running decommission. You can override it **per decommission** via the `timeout`
+> field/param of the worker `exit` API (see below) or the CLI `--exit-timeout` option, without
+> restarting the worker or editing configuration.
+
## Perform Decommissioning
@@ -63,6 +68,12 @@ Administrators perform decommissioning operation in two approaches:
```shell
curl -X POST -H "Content-Type: application/json" -d '{"type":"Decommission"}' http://ip:port/api/v1/workers/exit
```
+ You can override `celeborn.worker.decommission.forceExitTimeout` for this single decommission
+ by passing an optional `timeout` duration (e.g. `600s`/`30m`/`1h`); if omitted, the config
+ value (default `6h`) is used:
+ ```shell
+ curl -X POST -H "Content-Type: application/json" -d '{"type":"Decommission","timeout":"30m"}' http://ip:port/api/v1/workers/exit
+ ```
2. Via Celeborn Master(Leader) REST API endpoint:
```shell
curl -X POST -H "Content-Type: application/json" -d '{"eventType":"Decommission","workers":[{"host":"192.168.15.140","rpcPort":"37359","pushPort":"38303","fetchPort":"37569","replicatePort":"37093"},{"host":"192.168.15.141","rpcPort":"37359","pushPort":"38303","fetchPort":"37569","replicatePort":"37093"}]}' http://ip:port/api/v1/workers/events
diff --git a/docs/restapi.md b/docs/restapi.md
index 4f2892aec..fb2112ada 100644
--- a/docs/restapi.md
+++ b/docs/restapi.md
@@ -79,7 +79,7 @@ See the [migration guide](migration.md) for API mappings.
| /threadDump | GET | | List the current thread dump of the worker. |
| /unavailablePeers | GET | | List the unavailable peers of the worker, this always means the worker connect to the peer failed. |
| /workerInfo | GET | | List the worker information of the worker. |
-| /exit | POST | type=${EXIT_TYPE} | Trigger this worker to exit. Legal `type`s are 'Decommission', 'Graceful' and 'Immediately'. |
+| /exit | POST | type=${EXIT_TYPE}&timeout=${TIMEOUT} | Trigger this worker to exit. Legal `type`s are 'Decommission', 'Graceful' and 'Immediately'. Optional `timeout` (e.g. 600s/30m/1h) overrides `celeborn.worker.decommission.forceExitTimeout` for this single Decommission exit; ignored for other types. |
### `/api/v1` APIs (Since 0.6.0)
diff --git a/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/WorkerExitRequest.java b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/WorkerExitRequest.java
index 7530cb196..8e8d89172 100644
--- a/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/WorkerExitRequest.java
+++ b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/WorkerExitRequest.java
@@ -32,7 +32,8 @@ import com.fasterxml.jackson.annotation.JsonTypeName;
* WorkerExitRequest
*/
@JsonPropertyOrder({
- WorkerExitRequest.JSON_PROPERTY_TYPE
+ WorkerExitRequest.JSON_PROPERTY_TYPE,
+ WorkerExitRequest.JSON_PROPERTY_TIMEOUT
})
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", comments = "Generator version: 7.8.0")
public class WorkerExitRequest {
@@ -78,6 +79,9 @@ public class WorkerExitRequest {
public static final String JSON_PROPERTY_TYPE = "type";
private TypeEnum type = TypeEnum.NONE;
+ public static final String JSON_PROPERTY_TIMEOUT = "timeout";
+ private String timeout;
+
public WorkerExitRequest() {
}
@@ -106,6 +110,32 @@ public class WorkerExitRequest {
this.type = type;
}
+ public WorkerExitRequest timeout(String timeout) {
+
+ this.timeout = timeout;
+ return this;
+ }
+
+ /**
+ * Optional decommission forceExitTimeout override (e.g. 600s/30m/1h) for this single
+ * decommission; ignored for non-Decommission types.
+ * @return timeout
+ */
+ @javax.annotation.Nullable
+ @JsonProperty(JSON_PROPERTY_TIMEOUT)
+ @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS)
+
+ public String getTimeout() {
+ return timeout;
+ }
+
+
+ @JsonProperty(JSON_PROPERTY_TIMEOUT)
+ @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS)
+ public void setTimeout(String timeout) {
+ this.timeout = timeout;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -115,12 +145,13 @@ public class WorkerExitRequest {
return false;
}
WorkerExitRequest workerExitRequest = (WorkerExitRequest) o;
- return Objects.equals(this.type, workerExitRequest.type);
+ return Objects.equals(this.type, workerExitRequest.type) &&
+ Objects.equals(this.timeout, workerExitRequest.timeout);
}
@Override
public int hashCode() {
- return Objects.hash(type);
+ return Objects.hash(type, timeout);
}
@Override
@@ -128,6 +159,7 @@ public class WorkerExitRequest {
StringBuilder sb = new StringBuilder();
sb.append("class WorkerExitRequest {\n");
sb.append(" type: ").append(toIndentedString(type)).append("\n");
+ sb.append(" timeout: ").append(toIndentedString(timeout)).append("\n");
sb.append("}");
return sb.toString();
}
diff --git a/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml b/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml
index 1cb04ddcb..30b936bbe 100644
--- a/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml
+++ b/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml
@@ -759,6 +759,11 @@ components:
- GRACEFUL
- IMMEDIATELY
- NONE
+ timeout:
+ type: string
+ description: >-
+ Optional decommission forceExitTimeout override for this single decommission
+ (e.g. 600s/30m/1h); ignored for non-Decommission types.
LoggerInfo:
type: object
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
index a66f53305..a230b8860 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
@@ -183,7 +183,9 @@ abstract class HttpService extends Service with Logging {
def isDecommissioning: String = throw new UnsupportedOperationException()
- def exit(exitType: String): String = throw new UnsupportedOperationException()
+ def exit(exitType: String): String = exit(exitType, null)
+
+ def exit(exitType: String, timeout: String): String = throw new UnsupportedOperationException()
def handleWorkerEvent(
workerEventType: WorkerEventType,
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index 09999bad9..dd1066ad4 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -951,12 +951,35 @@ private[celeborn] class Worker(
sb.toString()
}
- override def exit(exitType: String): String = {
+ @volatile private var decommissionTimeoutOverrideMs: Long = -1L
+
+ override def exit(exitType: String, timeout: String): String = {
exitType.toUpperCase(Locale.ROOT) match {
case "DECOMMISSION" =>
- ShutdownHookManager.get().updateTimeout(
- conf.workerDecommissionForceExitTimeout,
- TimeUnit.MILLISECONDS)
+ val parsedTimeoutMs: Long =
+ if (timeout != null && timeout.nonEmpty) {
+ val ms = Utils.timeStringAsMs(timeout)
+ if (ms <= 0) {
+ logWarning(s"Invalid exit timeout '$timeout', falling back to config. " +
+ "Use a positive duration like 600s / 30m / 1h.")
+ -1L
+ } else {
+ ms
+ }
+ } else {
+ -1L
+ }
+ val fromApi = parsedTimeoutMs > 0
+ val effectiveTimeoutMs =
+ if (fromApi) {
+ parsedTimeoutMs
+ } else {
+ conf.workerDecommissionForceExitTimeout
+ }
+ decommissionTimeoutOverrideMs = effectiveTimeoutMs
+ logInfo(s"Worker decommission with forceExitTimeout=${effectiveTimeoutMs}ms " +
+ s"(source=${if (fromApi) "exit API" else "config"}).")
+ ShutdownHookManager.get().updateTimeout(effectiveTimeoutMs, TimeUnit.MILLISECONDS)
workerStatusManager.doTransition(WorkerEventType.Decommission)
case "GRACEFUL" =>
workerStatusManager.doTransition(WorkerEventType.Graceful)
@@ -1031,7 +1054,12 @@ private[celeborn] class Worker(
sendWorkerDecommissionToMaster()
shutdown.set(true)
val interval = conf.workerDecommissionCheckInterval
- val timeout = conf.workerDecommissionForceExitTimeout
+ val timeout =
+ if (decommissionTimeoutOverrideMs > 0) {
+ decommissionTimeoutOverrideMs
+ } else {
+ conf.workerDecommissionForceExitTimeout
+ }
var waitTimes = 0
def waitTime: Long = waitTimes * interval
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
index 087bc89e3..a2ec4e64c 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
@@ -79,13 +79,17 @@ class ApiWorkerResource extends ApiRequestContext {
@Path("/exit")
@Operation(description =
- "Trigger this worker to exit. Legal types are 'Decommission', 'Graceful' and 'Immediately'.")
+ "Trigger this worker to exit. Legal types are 'Decommission', 'Graceful' and 'Immediately'. " +
+ "Optional 'timeout' (e.g. 600s/30m/1h) overrides the decommission forceExitTimeout " +
+ "for this single decommission; ignored for non-Decommission types.")
@ApiResponse(
responseCode = "200",
content = Array(new Content(
mediaType = MediaType.APPLICATION_FORM_URLENCODED)))
@POST
- def exit(@FormParam("type") exitType: String): String = {
- httpService.exit(normalizeParam(exitType))
+ def exit(
+ @FormParam("type") exitType: String,
+ @FormParam("timeout") timeout: String): String = {
+ httpService.exit(normalizeParam(exitType), normalizeParam(timeout))
}
}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala
index 38a745a43..e21db8b4d 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala
@@ -72,7 +72,9 @@ class WorkerResource extends ApiRequestContext {
}
@Operation(description =
- "Trigger this worker to exit. Legal exit types are 'DECOMMISSION', 'GRACEFUL' and 'IMMEDIATELY'.")
+ "Trigger this worker to exit. Legal exit types are 'DECOMMISSION', 'GRACEFUL' and 'IMMEDIATELY'. " +
+ "Optional 'timeout' (e.g. 600s/30m/1h) overrides the decommission forceExitTimeout " +
+ "for this single decommission; ignored for non-Decommission types.")
@ApiResponse(
responseCode = "200",
content = Array(new Content(
@@ -83,6 +85,6 @@ class WorkerResource extends ApiRequestContext {
def exit(request: WorkerExitRequest): HandleResponse = {
new HandleResponse()
.success(true)
- .message(httpService.exit(request.getType.toString))
+ .message(httpService.exit(request.getType.toString, request.getTimeout))
}
}