Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3782947102
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -195,41 +205,41 @@ 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(5,
TimeUnit.SECONDS)).asTwitter()
+ }
+ }.toSeq
- Future.collect(gracefulStops).unit
- }
+ Future.collect(gracefulStopRequests).unit
+ }
+ .within(killTimeout)
Review Comment:
Taking the justification. This round answers the sizing question rather than
sidestepping it: `DefaultTerminationTimeoutMs` is now
`DefaultGracefulStopTimeoutMs + DefaultTerminationDrainAllowanceMs`
(`RegionExecutionManager.scala:82-86`) and the comment states the split — 5 s
for the stop, 1 s for the drain. A composite bound with a stated decomposition
is a defensible design; my objection was that the 6 s was an unexplained figure
that gracefulStop's 5 s deadline happened to sit inside, and it no longer is.
Resolving.
One thing the split does not carry over into the code, raised separately in
this round's review: `DefaultGracefulStopTimeoutMs` names gracefulStop's
deadline but does not set it — `:224` still spells it `ScalaDuration(5,
TimeUnit.SECONDS)`.
--
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]
