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 <[email protected]> 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)) } }
