SteNicholas commented on code in PR #3698:
URL: https://github.com/apache/celeborn/pull/3698#discussion_r3396619380
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -626,22 +626,25 @@ private[celeborn] class Worker(
jvmQuake.stop()
}
if (sendHeartbeatTask != null) {
- if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN) {
+ if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN ||
+ exitKind == CelebornExitKind.WORKER_DECOMMISSION) {
sendHeartbeatTask.cancel(false)
} else {
sendHeartbeatTask.cancel(true)
}
sendHeartbeatTask = null
}
if (checkFastFailTask != null) {
- if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN) {
+ if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN ||
+ exitKind == CelebornExitKind.WORKER_DECOMMISSION) {
checkFastFailTask.cancel(false)
} else {
checkFastFailTask.cancel(true)
}
checkFastFailTask = null
}
- if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN) {
+ if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN ||
+ exitKind == CelebornExitKind.WORKER_DECOMMISSION) {
Review Comment:
Grouping `WORKER_DECOMMISSION` with `WORKER_GRACEFUL_SHUTDOWN` here means
the thread pools now get an orderly `shutdown()` (drains in-flight tasks).
Before this PR the decommission path went through `stop(EXIT_IMMEDIATELY)` ->
`shutdownNow()`, so in-flight replicate/commit tasks (e.g. replicating to a
dead peer) are now drained rather than force-cancelled, and the wait is
deferred to `EXECUTOR.awaitTermination(workerGracefulShutdownTimeoutMs)` in
`shutdownExecutor`. Probably intended since decommission already waited for
shuffle to drain — but worth confirming, and note the decommission path now
silently depends on `workerGracefulShutdownTimeoutMs` even though the feature
disables graceful shutdown.
##########
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.workerDecommissionShutdown) {
+ ShutdownHookManager.get().addShutdownHook(
+ shutdownHookThread,
+ WORKER_SHUTDOWN_PRIORITY,
+ conf.workerDecommissionForceExitTimeout +
conf.workerDecommissionCheckInterval,
Review Comment:
The hook future is given `forceExitTimeout + checkInterval`, but
`decommissionWorker()`'s wait loop is bounded by `waitTime < forceExitTimeout`
with the bound checked *before* the sleep, so the loop alone can run up to
~`forceExitTimeout + checkInterval`. `stop(WORKER_DECOMMISSION)` then runs
*after* the loop and isn't cheap — each of the three `TransportServer.shutdown`
calls does `channel().close().awaitUninterruptibly(10s)`, plus flusher
shutdown. When shuffle never drains, the loop consumes the whole budget and
`executeShutdown` fires `future.cancel(true)` mid-`stop()`, truncating teardown.
The doc has the same gap: it tells operators to set
`terminationGracePeriodSeconds` to `forceExitTimeout + checkInterval`, so the
K8s SIGKILL lands at the same point. Consider sizing both the hook timeout and
the recommended grace period to include a `stop()` budget, or bounding the wait
loop strictly below `forceExitTimeout` to reserve headroom.
##########
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 {
Review Comment:
The hook matches `exitEventType` twice back-to-back — once to pick the
action (L1082) and once to pick the `stop()` exit-kind (L1091). Folding into a
single match (`case Decommission => decommissionWorker();
stop(WORKER_DECOMMISSION)`, etc.) keeps each event's action and exit-kind from
drifting apart when a new type is added.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -626,22 +626,25 @@ private[celeborn] class Worker(
jvmQuake.stop()
}
if (sendHeartbeatTask != null) {
- if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN) {
+ if (exitKind == CelebornExitKind.WORKER_GRACEFUL_SHUTDOWN ||
+ exitKind == CelebornExitKind.WORKER_DECOMMISSION) {
Review Comment:
`exitKind == WORKER_GRACEFUL_SHUTDOWN || exitKind == WORKER_DECOMMISSION` is
repeated three times in `stop()` (here, L639, L647). A single `val gracefulLike
= …` computed once would avoid the next exit-kind needing three coordinated
edits.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/worker/WorkerStatusManagerSuite.scala:
##########
@@ -57,24 +57,52 @@ class WorkerStatusManagerSuite extends AnyFunSuite {
statusManager.init(worker)
statusManager.doTransition(WorkerEventType.DecommissionThenIdle)
- Assert.assertEquals(statusManager.getWorkerState(),
PbWorkerStatus.State.InDecommissionThenIdle)
+ Assert.assertEquals(PbWorkerStatus.State.InDecommissionThenIdle,
statusManager.getWorkerState())
Assert.assertEquals(
- worker.workerInfo.getWorkerStatus().getStateValue,
- PbWorkerStatus.State.InDecommissionThenIdle.getNumber)
+ PbWorkerStatus.State.InDecommissionThenIdle.getNumber,
+ worker.workerInfo.getWorkerStatus().getStateValue)
// Rerun state Transition
statusManager.doTransition(WorkerEventType.DecommissionThenIdle)
- Assert.assertEquals(statusManager.getWorkerState(),
PbWorkerStatus.State.InDecommissionThenIdle)
+ Assert.assertEquals(PbWorkerStatus.State.InDecommissionThenIdle,
statusManager.getWorkerState())
// Reset shuffleKeys
shuffleKeys.clear()
statusManager.doTransition(WorkerEventType.DecommissionThenIdle)
- Assert.assertEquals(statusManager.getWorkerState(),
PbWorkerStatus.State.Idle)
+ Assert.assertEquals(PbWorkerStatus.State.Idle,
statusManager.getWorkerState())
statusManager.doTransition(WorkerEventType.Recommission)
- Assert.assertEquals(statusManager.getWorkerState(),
PbWorkerStatus.State.Normal)
+ Assert.assertEquals(PbWorkerStatus.State.Normal,
statusManager.getWorkerState())
statusManager.doTransition(WorkerEventType.Recommission)
- Assert.assertEquals(statusManager.getWorkerState(),
PbWorkerStatus.State.Normal)
+ Assert.assertEquals(PbWorkerStatus.State.Normal,
statusManager.getWorkerState())
+ }
+
+ test("Test exitEventType initialization based on config") {
+ // Default: neither graceful nor decommission → Immediately
+ val conf1 = new CelebornConf()
Review Comment:
`conf1`'s `assertEquals(Immediately, mgr1.exitEventType)` relies on no
`celeborn.worker.*.shutdown.enabled` system property being present; `new
CelebornConf()` loads sys-props, so leakage from another test/CI could flake
it. Setting the relevant keys explicitly (as conf2–conf4 do) keeps it hermetic.
Minor: the test never calls `init(worker)`, so only construction-time
`exitEventType` is covered.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -156,7 +156,7 @@ private[celeborn] class Worker(
private val WORKER_SHUTDOWN_PRIORITY = 100
val shutdown = new AtomicBoolean(false)
- private val gracefulShutdown = conf.workerGracefulShutdown
+ private val gracefulShutdown = conf.workerGracefulShutdown &&
!conf.workerDecommissionShutdown
Review Comment:
`conf.workerGracefulShutdown && !conf.workerDecommissionShutdown` is now
hand-copied in three places — here, `StorageManager.scala:298`, and
`PartitionFilesSorter.java:116`. Consider a single derived accessor on
`CelebornConf` (e.g. `effectiveWorkerGracefulShutdown`) so the override rule
can't drift across Scala/Java if the semantics ever change.
##########
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.workerDecommissionShutdown) {
Review Comment:
This `if/else` registers the hook twice just to vary the timeout. The
codebase already has the idiom: register once with the 2-arg overload and then
`ShutdownHookManager.get().updateTimeout(...)` — which is exactly what the REST
`exit("DECOMMISSION")` path does (`Worker.scala:955`).
--
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]
