Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3782964164
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -195,41 +210,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()
Review Comment:
I've accepted this round's answer to the shared-window thread — the 5s/1s
split at `:82-86`. But the split lives in constants and prose only: this line
still spells gracefulStop's deadline as its own literal, so
`DefaultGracefulStopTimeoutMs` names the value without setting it. Its only
reader is the sum at `:86`.
The cost is the next edit, in either direction. Raise the constant and only
the outer bound moves, so the "1s drain allowance" quietly becomes 6s. Raise
this literal and the outer bound stays at 6s, making gracefulStop's own
deadline unreachable — the defect this thread was about, back with nothing
saying so. Both edits look right at their own site, because the constant reads
as authoritative.
```suggestion
gracefulStop(
actorRef,
ScalaDuration(
RegionExecutionManager.DefaultGracefulStopTimeoutMs,
TimeUnit.MILLISECONDS
)
).asTwitter()
```
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala:
##########
@@ -79,7 +82,8 @@ object RegionExecutionManagerTestSupport {
case class CoordinatorHarnessFixture(
actorService: PekkoActorService,
- actorRefService: PekkoActorRefMappingService
+ actorRefService: PekkoActorRefMappingService,
+ coordinatorRef: ActorRef
Review Comment:
`coordinatorRef` is built at `:265` and assigned here, but none of the five
`createCoordinatorHarness` call sites read it — they take `.actorService` and
`.actorRefService` only. A fixture field that looks available but is plumbed
nowhere invites the next author to reach for the harness ref and find it leads
nowhere.
My miss from round 3, where I skipped this file as harness plumbing.
```suggestion
actorRefService: PekkoActorRefMappingService
```
(and drop `coordinatorRef = coordinatorRef` from the constructor call at
`:271`)
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala:
##########
@@ -185,7 +198,9 @@ class RegionExecutionManager(
}
private def terminateWorkers(regionExecution: RegionExecution) = {
- // 1. Send EndWorkers to every worker
+ implicit val timer: Timer = killRetryTimer
+ val killTimeout =
com.twitter.util.Duration.fromMilliseconds(terminationTimeoutMs)
+ // 1. Send EndWorkers with timeout
Review Comment:
The timeout isn't applied at step 1. `.within(killTimeout)` sits at `:230`,
outside the `flatMap`, so it bounds the EndWorker collect and the gracefulStop
collect together. That is what the shared-window thread was about, which makes
this the comment a future reader is most likely to act on.
```suggestion
// 1. Send EndWorker to every worker.
```
--
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]
