Copilot commented on code in PR #3828:
URL: https://github.com/apache/celeborn/pull/3828#discussion_r3904587304
##########
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))
}
Review Comment:
`timeout` is optional per the docs, but `@FormParam(\"timeout\")` will be
`null` when omitted and is passed into `normalizeParam(timeout)`. If
`normalizeParam` assumes non-null (common for trim/uppercasing), this can throw
at runtime. Also, normalizing `timeout` the same way as `type` can be
problematic (e.g., uppercasing) for duration strings; consider only trimming
`timeout` (or using `@DefaultValue(\"\")` / `Option[String]`) and avoid
applying exit-type normalization rules to it.
##########
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.
Review Comment:
The PR metadata checkbox indicates no user-facing change, but this PR adds a
new CLI flag (`--exit-timeout`) and a new REST API parameter (`timeout`) which
are user-facing. Please update the PR description checklist accordingly so
release notes / reviewers can correctly categorize the change.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -1031,7 +1064,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
+ }
Review Comment:
The override is stored in a shared mutable `@volatile` field and later read
to compute the decommission loop timeout. If multiple exit requests arrive
close together (or a second DECOMMISSION request is issued while the first is
in progress), the field can be overwritten and potentially affect the effective
timeout used by the in-flight decommission. Prefer associating the timeout
override with the specific decommission transition (e.g., pass it through the
transition/event payload or capture it into a local value used consistently by
the decommission routine) rather than using a process-wide mutable variable.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -951,12 +952,44 @@ private[celeborn] class Worker(
sb.toString()
}
- override def exit(exitType: String): String = {
+ @volatile private var decommissionTimeoutOverrideMs: Long = -1L
+
+ override def exit(exitType: String): String = exit(exitType, null)
+
+ override def exit(exitType: String, timeout: String): String = {
exitType.toUpperCase(Locale.ROOT) match {
case "DECOMMISSION" =>
- ShutdownHookManager.get().updateTimeout(
- conf.workerDecommissionForceExitTimeout,
- TimeUnit.MILLISECONDS)
+ val overrideMs: Long =
+ if (timeout != null && timeout.nonEmpty) {
+ try {
+ val ms = Utils.timeStringAsMs(timeout)
+ if (ms > 0) ms
+ else {
+ logWarning(s"Invalid exit timeout '$timeout', falling back to
config. " +
+ "Use a positive duration like 600s / 30m / 1h.")
+ -1L
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(
+ s"Unparseable exit timeout '$timeout', falling back to
config. " +
+ "Use a positive duration like 600s / 30m / 1h.",
+ e)
+ -1L
+ }
+ } else {
+ -1L
+ }
+ decommissionTimeoutOverrideMs = overrideMs
+ val effectiveTimeoutMs =
+ if (overrideMs > 0) {
+ overrideMs
+ } else {
+ conf.workerDecommissionForceExitTimeout
+ }
+ logInfo(s"Worker decommission with
forceExitTimeout=${effectiveTimeoutMs}ms " +
+ s"(source=${if (overrideMs > 0) "exit API" else "config"}).")
+ ShutdownHookManager.get().updateTimeout(effectiveTimeoutMs,
TimeUnit.MILLISECONDS)
workerStatusManager.doTransition(WorkerEventType.Decommission)
Review Comment:
The override is stored in a shared mutable `@volatile` field and later read
to compute the decommission loop timeout. If multiple exit requests arrive
close together (or a second DECOMMISSION request is issued while the first is
in progress), the field can be overwritten and potentially affect the effective
timeout used by the in-flight decommission. Prefer associating the timeout
override with the specific decommission transition (e.g., pass it through the
transition/event payload or capture it into a local value used consistently by
the decommission routine) rather than using a process-wide mutable variable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
