Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3755056591
##########
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:
This gets the mutation onto the right thread but drops the ordering the
inline placement had. The send is fire-and-forget and `Future.Unit` follows
immediately, so the termination future resolves while the message is still
queued.
Its continuation is `advanceRegionExecutions`
(WorkflowExecutionManager.scala:110-117), running on the global EC. That
rebuilds the region with the same worker ids (:140-153) and sends control RPCs,
which can take the old sequence counter while the fresh worker's channel starts
at 0. The late cleanup then renumbers into the same range — the silent loss
#6920 describes.
Keep the hop, restore the ordering: give the message a `Promise[Unit]` the
handler completes after the removals, and return it instead of `Future.Unit`.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala:
##########
@@ -164,6 +165,18 @@ object RegionExecutionManagerTestSupport {
override def initState(): Unit = ()
override def loadFromCheckpoint(chkpt: CheckpointState): Unit = ()
+
+ override def receive: Receive = {
+ case Coordinator.CleanupWorkerChannels(workerIds) =>
Review Comment:
This duplicates `Coordinator.scala:193-198` line for line. So the spec's
"Seed stale control channels to verify that successful termination removes
them" fixture checks the harness's copy, not the production handler — drop
`outputGateway.removeControlChannel` from `Coordinator` and the suite still
passes.
Lift the three removals into one helper on the `Coordinator` companion and
call it from both sites. The assertion then exercises the real code again.
The `case msg => super.receive(msg)` below has the same total-function
problem as the production site.
##########
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:
The catch-all makes `receive` total, so `isDefinedAt` is always true and
Pekko never calls `unhandled`. An unmatched message reaches this branch, fails
the inner match, and throws `MatchError` inside the actor.
That is reachable today: `ClusterListener` → `AmberClient.notifyNodeFailure`
→ `ClientActor.scala:155` forwards a `WorkflowRecoveryMessage` to the
Coordinator, which none of the three composed functions handle. `ClientActor`
sets no `supervisorStrategy`, so the default restarts the Coordinator — wiping
`cp` and re-running `initState()` mid-execution.
`WorkflowWorker.receive:137` keeps the `orElse` shape for exactly this
reason.
```suggestion
private def handleCleanupWorkerChannels: Receive = {
case Coordinator.CleanupWorkerChannels(workerIds) =>
workerIds.foreach { workerId =>
cp.asyncRPCClient.inputGateway.removeControlChannel(workerId)
cp.asyncRPCClient.outputGateway.removeControlChannel(workerId)
cp.actorRefService.removeActorRef(workerId)
}
}
override def receive: Receive =
handleCleanupWorkerChannels orElse
super.receive orElse
handleDirectInvocation orElse
handleReplayMessages
```
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala:
##########
@@ -234,10 +254,11 @@ class RegionExecutionManagerSpec
assert(fixture.rpcProbe.endWorkerCalls.size == fixture.workerIds.size * 2)
}
- it should "default to a bounded ~1.4s termination budget" in {
- // 4 attempts from a 200 ms base, doubling: 200 + 400 + 800 ms = ~1.4 s of
waiting, not the
- // former 150 x 200 ms (~30 s). This is the documented contract for how
long a stuck region
- // blocks before failing loudly; pin it so changes are deliberate.
+ it should "default to a bounded ~25.4s termination budget" in {
+ // 4 attempts from a 200 ms base, doubling: 200 + 400 + 800 ms = ~1.4 s of
backoff, plus
+ // a 6 s timeout per attempt. Worst-case teardown is now ~25.4 s. This is
the documented
+ // contract for how long a stuck region blocks before failing loudly; pin
it so changes
+ // are deliberate.
assert(RegionExecutionManager.DefaultMaxTerminationAttempts == 4)
assert(RegionExecutionManager.DefaultKillRetryBaseBackoffMs == 200L)
Review Comment:
This test calls itself the documented contract, so the figure and the
assertions both have to hold. Neither does. `.within(killTimeout)` wraps the
EndWorker collect and the gracefulStop collect separately, so an attempt bounds
two stages. And `DefaultTerminationTimeoutMs`, now the dominant term, is never
asserted.
```suggestion
it should "default to a bounded termination budget" in {
// 4 attempts from a 200 ms base, doubling: 200 + 400 + 800 ms = ~1.4 s
of backoff. Each attempt
// then bounds two stages: a 6 s timeout on the EndWorker collect, and
gracefulStop's own 5 s
// deadline on the stop collect -- ~11 s per attempt, ~45 s overall.
This is the documented
// contract for how long a stuck region blocks before failing loudly;
pin it so changes
// are deliberate.
assert(RegionExecutionManager.DefaultMaxTerminationAttempts == 4)
assert(RegionExecutionManager.DefaultKillRetryBaseBackoffMs == 200L)
assert(RegionExecutionManager.DefaultTerminationTimeoutMs == 6000L)
```
##########
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:
The two rationale comments that sat on these removals in
`RegionExecutionManager` were dropped. This is now the only place the logic
lives, so nothing records why the channels have to go — the part a future
reader is most likely to simplify away.
```suggestion
// Removing a worker's actorRef stops other actors from reaching it.
Restarted regions reuse
// actorId, so the control channels must go too: otherwise the coordinator
resumes the old
// control-message sequence numbers and the new worker discards its
commands as duplicates.
case class CleanupWorkerChannels(workerIds: Seq[ActorVirtualIdentity])
```
##########
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:
This landed on the blank line that separated the `org.apache` imports from
the `scala` block. Same in `RegionExecutionManager.scala:64` and
`RegionExecutionManagerTestSupport.scala:60`.
```suggestion
import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
```
--
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]
