You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/06/26 23:42:58 UTC
(texera) branch gh-readonly-queue/main/pr-5737-349685d82b92d8216dc0388ff4c4df0a3d0dea33 created (now 56f363adff)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5737-349685d82b92d8216dc0388ff4c4df0a3d0dea33
in repository https://gitbox.apache.org/repos/asf/texera.git
at 56f363adff fix(amber): bound region termination retry instead of looping forever (#5737)
This branch includes the following new commits:
new 56f363adff fix(amber): bound region termination retry instead of looping forever (#5737)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(amber): bound region termination retry instead of looping forever (#5737)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5737-349685d82b92d8216dc0388ff4c4df0a3d0dea33
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 56f363adff3af3378a8310c7885d02966f471ee3
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Fri Jun 26 16:42:50 2026 -0700
fix(amber): bound region termination retry instead of looping forever (#5737)
### What changes were proposed in this PR?
- `RegionExecutionCoordinator.terminateWorkersWithRetry` retried worker
termination with no limit, so a worker that never finished draining (its
`EndWorker` kept failing) left the region's termination future
unresolved and the workflow never reached COMPLETED, surfacing as an
opaque ~1 minute timeout in `DataProcessingSpec` with no indication of
the stuck region or workers.
- Bounded the retry by a new `maxTerminationAttempts` budget (default
150, about 30s at the existing 200ms delay); on exhaustion the
termination future fails with a descriptive error naming the region and
the still-unterminated workers, instead of retrying indefinitely.
- Made `maxTerminationAttempts` and `killRetryDelay` constructor
parameters with production defaults so the behavior is unit-testable
without long waits.
- Scope note: this is a fail-fast/diagnosability change (it converts a
silent hang into a fast, explicit failure, matching the pattern in
#4683), not a guaranteed elimination of the underlying termination
flake.
### Any related issues, documentation, discussions?
Closes: #5614
### How was this PR tested?
- Run `sbt "WorkflowExecutionService/testOnly
*RegionExecutionCoordinatorSpec"` (under JDK 17); expect 3 tests
succeeded, 0 failed.
- New test `give up with a descriptive error once the EndWorker retry
budget is exhausted`: forces `EndWorker` to always fail, then asserts
the completion future fails with `IllegalStateException` containing
"could not be terminated after 3 attempts", the coordinator is not
marked completed, exactly 3 `EndWorker` calls were made, and the worker
actor ref is retained.
- Existing tests in the same spec (gracefulStop ordering,
transient-failure retry-then-succeed) still pass, confirming the normal
and transient paths are unchanged.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8 in compliance with ASF
---
.../scheduling/RegionExecutionCoordinator.scala | 34 ++++-
.../scheduling/RegionCoordinatorTestSupport.scala | 29 +++-
.../RegionExecutionCoordinatorSpec.scala | 148 ++++++++++++++++++++-
3 files changed, 201 insertions(+), 10 deletions(-)
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
index aba6eff759..9e84e5e80b 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
@@ -64,6 +64,15 @@ import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
import scala.concurrent.duration.{Duration => ScalaDuration}
+object RegionExecutionCoordinator {
+
+ // Max EndWorker retries before termination fails. ~30s at DefaultKillRetryDelay (200ms).
+ private[scheduling] val DefaultMaxTerminationAttempts: Int = 150
+
+ private[scheduling] val DefaultKillRetryDelay: TwitterDuration =
+ TwitterDuration.fromMilliseconds(200)
+}
+
/**
* The executor of a region.
*
@@ -97,7 +106,9 @@ class RegionExecutionCoordinator(
asyncRPCClient: AsyncRPCClient,
controllerConfig: ControllerConfig,
actorService: PekkoActorService,
- actorRefService: PekkoActorRefMappingService
+ actorRefService: PekkoActorRefMappingService,
+ maxTerminationAttempts: Int = RegionExecutionCoordinator.DefaultMaxTerminationAttempts,
+ killRetryDelay: TwitterDuration = RegionExecutionCoordinator.DefaultKillRetryDelay
) extends AmberLogging {
initRegionExecution()
@@ -113,7 +124,6 @@ class RegionExecutionCoordinator(
)
private val terminationFutureRef: AtomicReference[Future[Unit]] = new AtomicReference(null)
private val killRetryTimer: Timer = new JavaTimer(true)
- private val killRetryDelay: TwitterDuration = TwitterDuration.fromMilliseconds(200)
/**
* Sync the status of `RegionExecution` and transition this coordinator's phase to `Completed` only when the
@@ -216,9 +226,27 @@ class RegionExecutionCoordinator(
attempt: Int = 1
): Future[Unit] = {
terminateWorkers(regionExecution).rescue {
+ case err if attempt >= maxTerminationAttempts =>
+ val workerIds = regionExecution.getAllOperatorExecutions.flatMap {
+ case (_, opExec) => opExec.getWorkerIds
+ }.toSeq
+ val attemptsLabel = if (attempt == 1) "1 attempt" else s"$attempt attempts"
+ logger.error(
+ s"Region ${region.id.id} could not be terminated after $attemptsLabel; giving up. " +
+ s"Workers still not terminated: ${workerIds.mkString(", ")}.",
+ err
+ )
+ Future.exception(
+ new IllegalStateException(
+ s"Region ${region.id.id} could not be terminated after $attemptsLabel " +
+ s"(workers still not terminated: ${workerIds.mkString(", ")}).",
+ err
+ )
+ )
case err =>
logger.warn(
- s"Failed to terminate region ${region.id.id} on attempt $attempt. Retrying in ${killRetryDelay.inMilliseconds} ms.",
+ s"Failed to terminate region ${region.id.id} on attempt $attempt of $maxTerminationAttempts. " +
+ s"Retrying in ${killRetryDelay.inMilliseconds} ms.",
err
)
Future
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
index 5673c02691..64d85972ff 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
@@ -176,10 +176,22 @@ object RegionCoordinatorTestSupport {
def createWorkerId(physicalOp: PhysicalOp): ActorVirtualIdentity =
VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, physicalOp.id, 0)
+ def createWorkerIds(physicalOp: PhysicalOp, count: Int): Seq[ActorVirtualIdentity] =
+ (0 until count).map(i =>
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, physicalOp.id, i)
+ )
+
def createSingleWorkerRegion(
regionId: Long,
physicalOp: PhysicalOp,
workerId: ActorVirtualIdentity
+ ): Region =
+ createWorkerRegion(regionId, physicalOp, Seq(workerId))
+
+ def createWorkerRegion(
+ regionId: Long,
+ physicalOp: PhysicalOp,
+ workerIds: Seq[ActorVirtualIdentity]
): Region =
Region(
RegionIdentity(regionId),
@@ -187,7 +199,8 @@ object RegionCoordinatorTestSupport {
physicalLinks = Set.empty,
resourceConfig = Some(
ResourceConfig(
- operatorConfigs = Map(physicalOp.id -> OperatorConfig(List(WorkerConfig(workerId))))
+ operatorConfigs =
+ Map(physicalOp.id -> OperatorConfig(workerIds.map(WorkerConfig(_)).toList))
)
)
)
@@ -197,13 +210,21 @@ object RegionCoordinatorTestSupport {
seedRegionId: Long,
physicalOp: PhysicalOp,
workerId: ActorVirtualIdentity
+ ): Unit =
+ seedReusableWorkerExecutions(workflowExecution, seedRegionId, physicalOp, Seq(workerId))
+
+ def seedReusableWorkerExecutions(
+ workflowExecution: WorkflowExecution,
+ seedRegionId: Long,
+ physicalOp: PhysicalOp,
+ workerIds: Seq[ActorVirtualIdentity]
): Unit = {
// RegionExecutionCoordinator skips real worker creation when an execution for this operator
// already exists.
- workflowExecution
- .initRegionExecution(createSingleWorkerRegion(seedRegionId, physicalOp, workerId))
+ val operatorExecution = workflowExecution
+ .initRegionExecution(createWorkerRegion(seedRegionId, physicalOp, workerIds))
.initOperatorExecution(physicalOp.id)
- .initWorkerExecution(workerId)
+ workerIds.foreach(operatorExecution.initWorkerExecution)
}
def await[T](future: Future[T]): T = Await.result(future, testTimeout)
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinatorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinatorSpec.scala
index 2c663612c4..2feec2362c 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinatorSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinatorSpec.scala
@@ -19,7 +19,7 @@
package org.apache.texera.amber.engine.architecture.scheduling
-import com.twitter.util.Future
+import com.twitter.util.{Duration => TwitterDuration, Future}
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.testkit.TestKit
import org.apache.texera.amber.core.storage.VFSURIFactory
@@ -136,6 +136,104 @@ class RegionExecutionCoordinatorSpec
assert(workerState(fixture) == WorkerState.TERMINATED)
}
+ it should "give up with a descriptive error once the EndWorker retry budget is exhausted" in {
+ // EndWorker always fails: a worker that never finishes draining.
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => Some(transientEndWorkerFailure),
+ maxTerminationAttempts = 3,
+ killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ )
+
+ launchRegion(fixture.coordinator)
+ val completion = requestRegionCompletion(fixture.coordinator)
+
+ val failure = intercept[IllegalStateException] {
+ await(completion)
+ }
+ assert(failure.getMessage.contains("could not be terminated after 3 attempts"))
+ assert(!fixture.coordinator.isCompleted)
+ assert(fixture.rpcProbe.endWorkerCalls.size == 3)
+ assert(fixture.actorRefService.hasActorRef(fixture.workerId))
+ }
+
+ it should "give up after a single attempt when the budget is one" in {
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => Some(transientEndWorkerFailure),
+ maxTerminationAttempts = 1,
+ killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ )
+
+ launchRegion(fixture.coordinator)
+ val completion = requestRegionCompletion(fixture.coordinator)
+
+ val failure = intercept[IllegalStateException] {
+ await(completion)
+ }
+ assert(failure.getMessage.contains("could not be terminated after 1 attempt"))
+ assert(failure.getMessage.contains(fixture.workerId.toString))
+ assert(fixture.rpcProbe.endWorkerCalls.size == 1)
+ }
+
+ it should "complete when EndWorker finally succeeds on the last permitted attempt" in {
+ // Fail every attempt but the last permitted one. The give-up branch only fires when an attempt
+ // both fails AND is the final one, so a success on attempt == budget must still complete the
+ // region rather than report it as un-terminable. This pins the off-by-one boundary.
+ val attempts = new atomic.AtomicInteger(0)
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ =>
+ if (attempts.incrementAndGet() < 2) {
+ Some(transientEndWorkerFailure)
+ } else {
+ Some(EmptyReturn())
+ },
+ maxTerminationAttempts = 2,
+ killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ )
+
+ launchRegion(fixture.coordinator)
+ val completion = requestRegionCompletion(fixture.coordinator)
+
+ await(completion)
+ assert(fixture.coordinator.isCompleted)
+ assert(fixture.rpcProbe.endWorkerCalls.size == 2)
+ assert(!fixture.actorRefService.hasActorRef(fixture.workerId))
+ assert(workerState(fixture) == WorkerState.TERMINATED)
+ }
+
+ it should "list every still-running worker and preserve the cause when giving up" in {
+ // A region with several workers, all of which never finish draining. The give-up error must
+ // name each stuck worker (so the user can act on it) and chain the underlying failure as cause.
+ val fixture = createMultiWorkerGiveUpFixture(
+ workerCount = 3,
+ maxTerminationAttempts = 2,
+ killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ )
+
+ launchRegion(fixture.coordinator)
+ val completion = requestRegionCompletion(fixture.coordinator)
+
+ val failure = intercept[IllegalStateException] {
+ await(completion)
+ }
+ assert(failure.getMessage.contains("could not be terminated after 2 attempts"))
+ fixture.workerIds.foreach { workerId =>
+ assert(failure.getMessage.contains(workerId.toString))
+ }
+ assert(failure.getCause != null)
+ assert(!fixture.coordinator.isCompleted)
+ // EndWorker is sent to every worker on every attempt.
+ assert(fixture.rpcProbe.endWorkerCalls.size == fixture.workerIds.size * 2)
+ }
+
+ it should "default to a bounded ~30s termination budget" in {
+ // 150 attempts * 200 ms ≈ 30 s. These defaults are the documented contract for how long a
+ // stuck region blocks before failing loudly; pin them so changes are deliberate.
+ assert(RegionExecutionCoordinator.DefaultMaxTerminationAttempts == 150)
+ assert(
+ RegionExecutionCoordinator.DefaultKillRetryDelay == TwitterDuration.fromMilliseconds(200)
+ )
+ }
+
it should "surface the underlying cause when an output port schema is unavailable" in {
// Reproduces issue #3546: when schema inference for an output port fails (e.g. because a
// dataset used by the workflow has not been shared with the running user), the port's
@@ -226,7 +324,9 @@ class RegionExecutionCoordinatorSpec
)
private def createSingleRegionFixture(
- endWorkerResponse: WorkerRpcCall => Option[ControlReturn]
+ endWorkerResponse: WorkerRpcCall => Option[ControlReturn],
+ maxTerminationAttempts: Int = RegionExecutionCoordinator.DefaultMaxTerminationAttempts,
+ killRetryDelay: TwitterDuration = RegionExecutionCoordinator.DefaultKillRetryDelay
): SingleRegionFixture = {
val physicalOp = createSourceOp("test-op")
val workerId = createWorkerId(physicalOp)
@@ -253,7 +353,9 @@ class RegionExecutionCoordinatorSpec
rpcProbe.asyncRPCClient,
ControllerConfig(None, None, None, None),
controller.actorService,
- controller.actorRefService
+ controller.actorRefService,
+ maxTerminationAttempts,
+ killRetryDelay
)
SingleRegionFixture(
@@ -267,6 +369,46 @@ class RegionExecutionCoordinatorSpec
)
}
+ private case class MultiWorkerFixture(
+ coordinator: RegionExecutionCoordinator,
+ rpcProbe: ControllerRpcProbe,
+ workerIds: Seq[ActorVirtualIdentity]
+ )
+
+ // A region whose workers all fail EndWorker forever, used to exercise the give-up path's
+ // aggregation over multiple workers.
+ private def createMultiWorkerGiveUpFixture(
+ workerCount: Int,
+ maxTerminationAttempts: Int,
+ killRetryDelay: TwitterDuration
+ ): MultiWorkerFixture = {
+ val physicalOp = createSourceOp("multi-op")
+ val workerIds = createWorkerIds(physicalOp, workerCount)
+ val region = createWorkerRegion(1, physicalOp, workerIds)
+
+ val workflowExecution = WorkflowExecution()
+ seedReusableWorkerExecutions(workflowExecution, seedRegionId = 0, physicalOp, workerIds)
+ workflowExecution.initRegionExecution(region)
+
+ val rpcProbe = new ControllerRpcProbe(_ => Some(transientEndWorkerFailure))
+ val controller = createControllerHarness()
+ workerIds.foreach(registerLiveWorker(controller.actorRefService, _))
+
+ val coordinator = new RegionExecutionCoordinator(
+ region,
+ isRestart = false,
+ workflowExecution,
+ rpcProbe.asyncRPCClient,
+ ControllerConfig(None, None, None, None),
+ controller.actorService,
+ controller.actorRefService,
+ maxTerminationAttempts,
+ killRetryDelay
+ )
+
+ MultiWorkerFixture(coordinator, rpcProbe, workerIds)
+ }
+
private def launchRegion(coordinator: RegionExecutionCoordinator): Unit = {
await(coordinator.syncStatusAndTransitionRegionExecutionPhase())
}