Xiao-zhen-Liu commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r4200293676
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -195,41 +206,47 @@ class RegionExecutionManager(
}
}.toSeq
- val endWorkerFuture: Future[Unit] =
- Future.collect(endWorkerRequests).unit
-
- // 2. Send GracefulStops only after 1 has finished
- 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
+ val terminationAttempt =
+ Future
+ .collect(endWorkerRequests)
+ .unit
+ .flatMap { _ =>
+ // 2. Only send GracefulStops after all EndWorkers have succeeded.
+ val gracefulStopRequests =
+ regionExecution.getAllOperatorExecutions.flatMap {
+ case (_, opExec) =>
+ opExec.getWorkerIds.map { workerId =>
+ val actorRef = actorRefService.getActorRef(workerId)
+ gracefulStop(
+ actorRef,
+ ScalaDuration(
+ RegionExecutionManager.DefaultGracefulStopTimeoutMs,
+ TimeUnit.MILLISECONDS
+ )
+ ).asTwitter()
+ }
+ }.toSeq
- Future.collect(gracefulStops).unit
- }
+ Future.collect(gracefulStopRequests).unit
+ }
+ .within(killTimeout)
- // 3. Log whether the kills were successful
- gracefulStopRequests.transform {
+ // 3. Clean up only after graceful termination succeeds.
+ terminationAttempt.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()
}
}
- Future.Unit // propagate success
+ val cleanupPromise = Promise[Unit]()
+ actorService.self ! Coordinator.CleanupWorkerChannels(allWorkerIds,
cleanupPromise)
Review Comment:
`forceTerminate()` (line 244) still runs off the coordinator thread, so it
can race with `updateState` and set a TERMINATED worker back to RUNNING. It
should run in this message's handler too.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala:
##########
@@ -242,8 +256,10 @@ object RegionExecutionManagerTestSupport {
trait RegionExecutionManagerTestSupport { self: TestKit =>
import RegionExecutionManagerTestSupport._
- protected def createCoordinatorHarness(): CoordinatorHarnessFixture = {
- val coordinatorRef = TestActorRef(new CoordinatorHarness)
+ protected def createCoordinatorHarness(
+ rpcProbe: CoordinatorRpcProbe = new CoordinatorRpcProbe(_ => None)
+ ): CoordinatorHarnessFixture = {
+ val coordinatorRef = TestActorRef(new CoordinatorHarness(rpcProbe))
Review Comment:
`TestActorRef` handles this message synchronously, so the tests pass the
same with the fire-and-forget version. A test that holds the cleanup message
and checks that the region is not completed until it is handled would cover
this.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala:
##########
@@ -141,6 +141,26 @@ class RegionExecutionManagerSpec
assert(workerState(fixture) == WorkerState.TERMINATED)
}
+ it should "retry EndWorker when the termination timeout expires" in {
Review Comment:
Two cases have no test: a timeout followed by a successful retry, and a
failed gracefulStop leaving the refs for the next attempt. Please add both.
--
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]
