Copilot commented on code in PR #3828:
URL: https://github.com/apache/celeborn/pull/3828#discussion_r3862037105
##########
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
+ }
Review Comment:
`Utils.timeStringAsMs(timeout)` can throw (e.g., malformed duration like
`"abc"`), which would fail the exit request instead of falling back to config.
Wrap parsing in a `NonFatal` catch (or equivalent) and treat parse failures the
same as invalid/<=0 values (log + fallback).
##########
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
Review Comment:
`decommissionTimeoutOverrideMs` is intended as a per-decommission override,
but this assigns it even when no override was provided (it stores the config
value). Because the field is global state, a prior API-driven override can leak
into future decommissions that are triggered via other paths (e.g., master
event transition) and unexpectedly change their timeout. Consider storing only
the actual override (set to `-1` unless the API provided a valid timeout),
and/or resetting the override after the decommission completes or right before
starting decommission from a non-exit path.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -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
+ }
Review Comment:
With the current assignment behavior, `decommissionTimeoutOverrideMs` will
almost always be `> 0` after any decommission exit call (even when no override
is provided), making this conditional effectively redundant and obscuring the
intent. If you change the override field to represent “override-only”
(suggested above), this conditional becomes meaningful; otherwise consider
renaming the field to reflect that it stores the effective timeout rather than
an override.
##########
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()
Review Comment:
The new overload used by the worker endpoints defaults to throwing
`UnsupportedOperationException()`, which is risky if any `HttpService`
implementation doesn’t override the new method (existing overrides of
`exit(exitType)` won’t help for call sites now using `exit(exitType,
timeout)`). A safer backwards-compatible pattern is to make the new overload
delegate to the old method by default (ignoring `timeout`) and let
implementations override it only when they support the new behavior.
##########
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:
Since `timeout` is documented as optional, it’s better to model it as
optional at the JAX-RS boundary to avoid relying on `normalizeParam` for null
handling. Consider adding `@DefaultValue("")` for `timeout` (or using
`Option[String]` if that’s the established pattern here) so absent form params
don’t become `null` inputs downstream.
--
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]
