Yicong-Huang commented on code in PR #7122:
URL: https://github.com/apache/texera/pull/7122#discussion_r3782945782
##########
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:
Verified: fixed. The rationale at `RegionExecutionManagerSpec.scala:258-261`
now states one 6 s timeout per attempt and `~25.4 s` worst case, matching the
class comment at `RegionExecutionManager.scala:74-77`, and `4 x 6000 +
(200+400+800) = 25400 ms` checks out against the three constants the test pins
at `:262-264`. The two-stage figure is gone from both places. Resolving.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala:
##########
@@ -141,6 +141,45 @@ class RegionExecutionManagerSpec
assert(workerState(fixture) == WorkerState.TERMINATED)
}
+ it should "retry EndWorker when the termination timeout expires" in {
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => None,
+ maxTerminationAttempts = 2,
+ killRetryBaseBackoffMs = fastRetryBackoffMs,
+ terminationTimeoutMs = 10L
+ )
+
+ launchRegion(fixture.manager)
+ val completion = requestRegionCompletion(fixture.manager)
+
+ val failure = intercept[IllegalStateException] {
+ await(completion)
+ }
+
+ assert(failure.getMessage.contains("could not be terminated after 2
attempts"))
+ assert(fixture.rpcProbe.endWorkerCalls.size == 2)
+ assert(!fixture.manager.isCompleted)
+ }
+
+ it should "clean up control channels and actor refs after successful
termination" in {
Review Comment:
Verified: fixed by deletion, and nothing was lost with it.
`assertControlChannelsAreRemoved` is still invoked at `:104`, inside `"send
gracefulStop only after EndWorker succeeds"`, which also asserts `isCompleted`
(`:101`) and `!hasActorRef` (`:102`) — every assertion the removed test made.
Resolving.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala:
##########
@@ -251,7 +293,8 @@ class RegionExecutionManagerSpec
Time.withCurrentTimeFrozen { _ =>
val fixture = createSingleRegionFixture(
endWorkerResponse = _ => Some(transientEndWorkerFailure),
- killRetryTimer = timer
+ killRetryTimer = timer,
+ terminationTimeoutMs = Long.MaxValue
Review Comment:
Verified: fixed. Both call sites (`:277`, `:303`) now carry `// Disable the
termination timeout for this frozen-time retry test.`, which names the intent a
reader needs before `Long.MaxValue` reads as arbitrary. Resolving.
##########
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)
- // 3. Log whether the kills were successful
- gracefulStopRequests.transform {
+ // 3. Cleanup only after graceful termination succeeds.
Review Comment:
Verified: applied at `:232`. Resolving.
--
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]
