Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3741170063
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -196,37 +198,43 @@ class RegionExecutionManager(
}.toSeq
val endWorkerFuture: Future[Unit] =
- Future.collect(endWorkerRequests).unit
+ Future
+ .collect(endWorkerRequests)
+ .within(killTimeout)
+ .unit
- // 2. Send GracefulStops only after 1 has finished
+ // 2. Send GracefulStops with timeout
val gracefulStopRequests: Future[Unit] =
endWorkerFuture.flatMap { _ =>
val gracefulStops =
regionExecution.getAllOperatorExecutions.flatMap {
case (_, opExec) =>
opExec.getWorkerIds.map { workerId =>
val actorRef = actorRefService.getActorRef(workerId)
- // Remove the actorRef so that no other actors can find the
worker and send messages.
- actorRefService.removeActorRef(workerId)
- // Restarted regions reuse actorId. Remove stale control
channels so the
- // coordinator does not reuse old control-message sequence
numbers for new workers.
- asyncRPCClient.inputGateway.removeControlChannel(workerId)
- asyncRPCClient.outputGateway.removeControlChannel(workerId)
gracefulStop(actorRef, ScalaDuration(5,
TimeUnit.SECONDS)).asTwitter()
}
}.toSeq
- Future.collect(gracefulStops).unit
+ Future
+ .collect(gracefulStops)
+ .within(killTimeout)
+ .unit
}
- // 3. Log whether the kills were successful
+ // 3. Cleanup only after all gracefulStops succeed
gracefulStopRequests.transform {
case Return(_) =>
logger.debug(s"Region ${region.id.id} successfully terminated.")
regionExecution.getAllOperatorExecutions.foreach {
case (_, opExec) =>
opExec.getWorkerIds.foreach { workerId =>
opExec.getWorkerExecution(workerId).forceTerminate()
+ // Remove the actorRef after successful termination so other
actors cannot reach the worker.
+ actorRefService.removeActorRef(workerId)
+ // Restarted regions reuse actorId. Remove stale control
channels so the
+ // coordinator does not reuse old control-message sequence
numbers for new workers.
+ asyncRPCClient.inputGateway.removeControlChannel(workerId)
Review Comment:
These two removals used to run on the coordinator's actor thread: they sat
inside `endWorkerFuture.flatMap`, and the `EndWorker` reply is fulfilled while
the coordinator processes it. They now run in the `gracefulStop` continuation,
which `FutureBijection.asTwitter` completes on
`scala.concurrent.ExecutionContext.Implicits.global` (FutureBijection.scala:24,
44-51).
Both gateways hold plain `mutable.HashMap`s (`NetworkInputGateway.scala:34`,
`NetworkOutputGateway.scala:47`) that the actor thread reads while other
regions are still running (`Coordinator.scala:145,180`,
`WorkflowActor.scala:224`), so this is an unsynchronized cross-thread mutation.
Could the cleanup hop back onto the coordinator thread instead — the way
`PortCompletedHandler` defers to a later coordinator round? (`removeActorRef`
on the line above is already cross-thread via the 30 s resend timer, so it is
not part of this.)
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -185,7 +185,9 @@ class RegionExecutionManager(
}
private def terminateWorkers(regionExecution: RegionExecution) = {
- // 1. Send EndWorkers to every worker
+ implicit val timer: Timer = new JavaTimer(true)
+ val killTimeout = com.twitter.util.Duration.fromMilliseconds(1000)
Review Comment:
`killTimeout` is 1000 ms, but it wraps a `gracefulStop` that carries its own
5 s deadline (:214) — so that deadline can never be reached, and a worker that
stops in 1-5 s becomes a termination failure instead of a success.
The retry cannot recover it. Attempt 1 has already sent the stop message, so
attempt N+1's `EndWorker` reaches a worker that is stopping or gone and never
answers; each remaining attempt times out the same way and the region ends in
`IllegalStateException`. A Python proxy worker whose PVM takes ~2 s to exit —
the case #6920 and the description both name — used to terminate on the first
attempt.
Worth deriving both bounds from one constant so the outer `.within` stays
strictly greater than the inner deadline. (The thread above says the timeout
became 5 s; the shipped value is 1000 ms.)
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -196,37 +198,43 @@ class RegionExecutionManager(
}.toSeq
val endWorkerFuture: Future[Unit] =
- Future.collect(endWorkerRequests).unit
+ Future
+ .collect(endWorkerRequests)
+ .within(killTimeout)
+ .unit
- // 2. Send GracefulStops only after 1 has finished
+ // 2. Send GracefulStops with timeout
val gracefulStopRequests: Future[Unit] =
endWorkerFuture.flatMap { _ =>
val gracefulStops =
regionExecution.getAllOperatorExecutions.flatMap {
case (_, opExec) =>
opExec.getWorkerIds.map { workerId =>
val actorRef = actorRefService.getActorRef(workerId)
- // Remove the actorRef so that no other actors can find the
worker and send messages.
- actorRefService.removeActorRef(workerId)
- // Restarted regions reuse actorId. Remove stale control
channels so the
- // coordinator does not reuse old control-message sequence
numbers for new workers.
- asyncRPCClient.inputGateway.removeControlChannel(workerId)
- asyncRPCClient.outputGateway.removeControlChannel(workerId)
gracefulStop(actorRef, ScalaDuration(5,
TimeUnit.SECONDS)).asTwitter()
}
}.toSeq
- Future.collect(gracefulStops).unit
+ Future
+ .collect(gracefulStops)
+ .within(killTimeout)
Review Comment:
With two 1 s legs per attempt, worst-case teardown is now ~1.4 s of backoff
plus up to ~8 s of timeouts. The companion comment (:67-71) still says "200 +
400 + 800 ms = 1.4s in the worst case", and `RegionExecutionManagerSpec`'s
"default to a bounded ~1.4s termination budget" still calls that "the
documented contract for how long a stuck region blocks before failing loudly".
#7088 landed that bound one commit before this branch merged main, so
leaving both in place tells the next reader a budget the class no longer
honors. Please update the comment and the spec's rationale to state the real
one, timeout legs included.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -185,7 +185,9 @@ class RegionExecutionManager(
}
private def terminateWorkers(regionExecution: RegionExecution) = {
- // 1. Send EndWorkers to every worker
+ implicit val timer: Timer = new JavaTimer(true)
Review Comment:
This class already takes a `Timer` for exactly this concern —
`killRetryTimer` (:115) — and its sibling termination knobs are companion
defaults injected through the constructor (:72-74, :113-114). A fresh
`JavaTimer` here plus the hardcoded value below breaks that pattern where the
tests feel it: `RegionExecutionManagerSpec` drives every other knob through
`createSingleRegionFixture(killRetryTimer = ...)` and
`Time.withCurrentTimeFrozen`, but cannot control this one. The specs that hold
`endWorker` pending across the 5 s `testTimeout` now race a real 1 s clock, so
a loaded runner can turn them into a spurious retry-and-fail.
Reuse `killRetryTimer` as the implicit timer, and hoist the timeout to a
constructor param with a companion default alongside the other two.
--
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]
