Copilot commented on code in PR #3698:
URL: https://github.com/apache/celeborn/pull/3698#discussion_r3298950850
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -1072,29 +1075,42 @@ private[celeborn] class Worker(
workerStatusManager.transitionState(State.Exit)
}
- ShutdownHookManager.get().addShutdownHook(
- ThreadUtils.newThread(
- new Runnable {
- override def run(): Unit = {
- logInfo("Shutdown hook called.")
- workerStatusManager.exitEventType match {
- case WorkerEventType.Graceful =>
- shutdownGracefully()
- case WorkerEventType.Decommission =>
- decommissionWorker()
- case _ =>
- exitImmediately()
- }
+ private val shutdownHookThread = ThreadUtils.newThread(
+ new Runnable {
+ override def run(): Unit = {
+ logInfo("Shutdown hook called.")
+ workerStatusManager.exitEventType match {
+ case WorkerEventType.Graceful =>
+ shutdownGracefully()
+ case WorkerEventType.Decommission =>
+ decommissionWorker()
+ case _ =>
+ exitImmediately()
+ }
- if (workerStatusManager.exitEventType == WorkerEventType.Graceful) {
+ workerStatusManager.exitEventType match {
+ case WorkerEventType.Graceful =>
stop(CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN)
- } else {
+ case WorkerEventType.Decommission =>
+ stop(CelebornExitKind.WORKER_DECOMMISSION)
+ case _ =>
stop(CelebornExitKind.EXIT_IMMEDIATELY)
- }
}
- },
- "worker-shutdown-hook-thread"),
- WORKER_SHUTDOWN_PRIORITY)
+ }
+ },
+ "worker-shutdown-hook-thread")
+
+ if (conf.workerDecommissionShutdownEnabled) {
+ ShutdownHookManager.get().addShutdownHook(
+ shutdownHookThread,
+ WORKER_SHUTDOWN_PRIORITY,
+ conf.workerDecommissionForceExitTimeout +
conf.workerDecommissionCheckInterval,
Review Comment:
The shutdown hook timeout for decommission is set to `forceExitTimeout +
checkInterval`, but the PR description/docs suggest operators align
`forceExitTimeout` with the platform termination grace period. This mismatch
can cause the JVM to be SIGKILLed before the hook finishes. Either use
`forceExitTimeout` directly here (and adjust the decommission loop to respect
it strictly) or update the docs to recommend a grace period of at least
`forceExitTimeout + checkInterval` (plus some buffer for RPC/cleanup).
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1357,7 +1357,9 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
// //////////////////////////////////////////////////////
// Graceful Shutdown & Recover //
// //////////////////////////////////////////////////////
- def workerGracefulShutdown: Boolean = get(WORKER_GRACEFUL_SHUTDOWN_ENABLED)
+ def workerDecommissionShutdownEnabled: Boolean =
get(WORKER_DECOMMISSION_SHUTDOWN_ENABLED)
Review Comment:
`workerGracefulShutdown` was removed/renamed to
`workerGracefulShutdownEnabled`. Since `CelebornConf` is a public API used
across modules (and potentially by external code), removing the old accessor is
a source/binary compatibility break. Consider keeping `workerGracefulShutdown`
as a deprecated alias delegating to `workerGracefulShutdownEnabled` to preserve
compatibility while introducing the new override semantics.
##########
docs/configuration/worker.md:
##########
@@ -78,6 +78,7 @@ license: |
| celeborn.worker.congestionControl.workerProduceSpeed.low.watermark |
9223372036854775807b | false | Stop congestion If worker total produce speed
less than this configuration | 0.6.0 | |
| celeborn.worker.decommission.checkInterval | 30s | false | The wait interval
of checking whether all the shuffle expired during worker decommission | 0.4.0
| |
| celeborn.worker.decommission.forceExitTimeout | 6h | false | The wait time
of waiting for all the shuffle expire during worker decommission. | 0.4.0 | |
+| celeborn.worker.decommission.shutdown.enabled | false | false | When true,
the worker will decommission on shutdown signal (e.g. SIGTERM), waiting for all
shuffle data to be consumed or expired before exiting. This is suitable for
permanent scale-down scenarios where the worker will not restart. When enabled,
this overrides celeborn.worker.graceful.shutdown.enabled (recovery state will
not be saved since the worker is not expected to come back). Operators should
align celeborn.worker.decommission.forceExitTimeout with the pod's
terminationGracePeriodSeconds. | 0.6.0 | |
Review Comment:
This doc tells operators to align
`celeborn.worker.decommission.forceExitTimeout` with
`terminationGracePeriodSeconds`, but the worker’s shutdown hook timeout is
`forceExitTimeout + checkInterval` (see Worker.scala). As written, following
this guidance can leave insufficient time for the decommission shutdown hook to
complete. Please update the guidance to reflect the actual timeout used (or
change the code to match the documented behavior).
--
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]
