Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3755030978
##########
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:
Verified: `DefaultTerminationTimeoutMs` is 6000 ms
(`RegionExecutionManager.scala:77`), strictly greater than the 5 s
`gracefulStop` deadline at `:218`, so a worker that stops in 1-5 s no longer
fails the region. Resolving.
##########
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:
Verified: `terminationTimeoutMs` is now a constructor param with a companion
default (`RegionExecutionManager.scala:77`, `:119`) and `implicit val timer:
Timer = killRetryTimer` (`:192`) reuses the injected timer, so the specs drive
both knobs. Resolving.
##########
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:
Verified: the three removals now run in `Coordinator.receive` on the actor
thread (`Coordinator.scala:193-198`), reached via `actorService.self !` — the
hop I asked about. Resolving this one; the *ordering* consequence of making it
a mailbox message is a separate new finding in this round.
--
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]
