Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3773317093
##########
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:
Verified: the class comment (`RegionExecutionManager.scala:74-76`) now
matches the code. The two per-stage `.within` calls were collapsed into one at
`:225`, so an attempt really is bounded by a single 6 s window and `4 x 6 s +
1.4 s = ~25.4 s` is the honest figure. Resolving this thread.
Two follow-ups live elsewhere in this round: the spec copy of the budget
still states the old two-stage figure (thread below), and collapsing to one
window is itself a behavior change I have flagged separately.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -196,39 +202,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.")
+ val allWorkerIds =
regionExecution.getAllOperatorExecutions.toSeq.flatMap {
+ case (_, opExec) => opExec.getWorkerIds
+ }
regionExecution.getAllOperatorExecutions.foreach {
case (_, opExec) =>
opExec.getWorkerIds.foreach { workerId =>
opExec.getWorkerExecution(workerId).forceTerminate()
}
}
+ actorService.self ! Coordinator.CleanupWorkerChannels(allWorkerIds)
Review Comment:
Verified: `terminateWorkers` now creates a `Promise[Unit]`, ships it on the
message, and returns it (`RegionExecutionManager.scala:240-242`);
`Coordinator.handleCleanupWorkerChannels` completes it after the removals
(`Coordinator.scala:215-222`). The termination future therefore resolves on the
coordinator actor thread with the channels already gone, so
`advanceRegionExecutions` cannot rebuild the region on a stale sequence
counter. Resolving.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Coordinator.scala:
##########
@@ -190,7 +190,15 @@ class Coordinator(
}
override def receive: Receive = {
- super.receive orElse handleDirectInvocation orElse handleReplayMessages
+ case Coordinator.CleanupWorkerChannels(workerIds) =>
+ workerIds.foreach { workerId =>
+ cp.asyncRPCClient.inputGateway.removeControlChannel(workerId)
+ cp.asyncRPCClient.outputGateway.removeControlChannel(workerId)
+ cp.actorRefService.removeActorRef(workerId)
+ }
+
+ case msg =>
+ (super.receive orElse handleDirectInvocation orElse
handleReplayMessages)(msg)
}
Review Comment:
Verified: `handleCleanupWorkerChannels` (`Coordinator.scala:215-222`)
matches only `CleanupWorkerChannels`, and `receive` (`:225-229`) keeps the
`orElse` chain, so `isDefinedAt` is false for anything unmatched and Pekko
routes it to `unhandled`. The `WorkflowRecoveryMessage` path from `ClientActor`
no longer throws. Resolving.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Coordinator.scala:
##########
@@ -66,7 +66,7 @@ final case class CoordinatorConfig(
)
object Coordinator {
-
+ case class CleanupWorkerChannels(workerIds: Seq[ActorVirtualIdentity])
Review Comment:
Verified: the rationale now sits on the message declaration
(`Coordinator.scala:76-79`) and states why the control channels have to go, not
just the actorRef. Resolving.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Coordinator.scala:
##########
@@ -44,7 +44,7 @@ import org.apache.texera.amber.engine.common.ambermessage.{
import org.apache.texera.amber.engine.common.virtualidentity.util.{CLIENT,
COORDINATOR, SELF}
import org.apache.texera.amber.engine.common.{CheckpointState, SerializedState}
import org.apache.texera.web.SessionState
-
+import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
Review Comment:
Verified: the stray blank line is gone. The import blocks were restyled more
broadly than this asked for, which I have raised as a separate scope note in
the review body. Resolving this thread.
--
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]
