Copilot commented on code in PR #3828:
URL: https://github.com/apache/celeborn/pull/3828#discussion_r3889522491
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -951,12 +952,42 @@ 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(_) =>
+ logWarning(s"Unparseable exit timeout '$timeout', falling back
to config. " +
+ "Use a positive duration like 600s / 30m / 1h.")
+ -1L
Review Comment:
The NonFatal parse failure path logs only a generic warning and drops the
underlying exception details, which makes it harder to diagnose why a provided
timeout was rejected (e.g., invalid suffix). Consider logging the exception as
well (Logging supports a throwable overload).
##########
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:
This endpoint now accepts and forwards an optional `timeout` form parameter.
There are existing REST API tests for other worker endpoints, but none covering
the `/exit` API surface; adding a focused test (e.g., using a stubbed
HttpService/Worker that does not actually terminate) would help prevent
regressions in parameter wiring and optional/empty handling.
--
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]
