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/07/30 03:43:05 UTC
(texera) branch gh-readonly-queue/main/pr-7088-4e152e0d823224a42a59ade8abbd1f2c9075ca11 created (now 37ea97d5b9)
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-7088-4e152e0d823224a42a59ade8abbd1f2c9075ca11
in repository https://gitbox.apache.org/repos/asf/texera.git
at 37ea97d5b9 chore(amber): bound region termination to a short backoff (#7088)
This branch includes the following new commits:
new 37ea97d5b9 chore(amber): bound region termination to a short backoff (#7088)
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: chore(amber): bound region termination to a short backoff (#7088)
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-7088-4e152e0d823224a42a59ade8abbd1f2c9075ca11
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 37ea97d5b9572499b3a7022dc1e621db1cc6c69d
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Wed Jul 29 20:43:00 2026 -0700
chore(amber): bound region termination to a short backoff (#7088)
### What changes were proposed in this PR?
**Root cause.** `RegionExecutionManager` retried region termination
(`EndWorker`, then `gracefulStop`) up to 150 times at a flat 200 ms
delay, so a region whose workers do not terminate held teardown for ~30
s before the failure surfaced. Killing a worker is deterministic --
`EndWorker` plus `gracefulStop` either succeed, or something upstream is
broken in a way retrying cannot fix (#6891, #6918). The long poll bought
nothing except a quiet ~30 s stall that masked the real bug.
```
Before: EndWorker fails -> 149 retries x 200 ms -> ~30 s of silence -> loud failure
After: EndWorker fails -> 3 retries (200/400/800 ms) -> ~1.4 s -> loud failure
```
| | Before | After |
| --- | --- | --- |
| Attempts | 150 | 4 (1 + 3 retries) |
| Delay between attempts | flat 200 ms | 200 -> 400 -> 800 ms |
| Worst-case wait per region | ~30 s | ~1.4 s |
| Give-up behavior | `IllegalStateException` naming stuck workers |
unchanged |
**The retry itself lives in a util, not in the manager** (per review):
`Utils.retry` keeps its
`attempts` + `baseBackoffTimeInMS`-doubling contract but now schedules
its waits on a `Timer`
instead of calling `Thread.sleep`, so it can be used from an actor or
coordinator thread -- a
blocking sleep there stalls every other message queued on that thread,
which is why this path could
not reuse the old version. An `onRetry` hook keeps caller context in the
log (the manager still
names the region and the attempt); callers that do not care can omit it.
```scala
Utils.retry(attempts, baseBackoffTimeInMS, timer, onRetry) { operationReturningAFuture }
```
The previous blocking `retry` had no production callers -- only its own
tests -- so it was replaced
rather than duplicated. A blocking front door can return as a thin
sibling if a caller needs one.
`RegionExecutionManager` is now just a configuration of that util (4
attempts from a 200 ms base)
plus its own give-up message. Two more copies of this loop live
elsewhere --
`LakeFSStorageClient.retryWithBackoff` and `FileService.awaitDependency`
-- but they sit in
`common/workflow-core` and `file-service`, neither of which can see
amber's `Utils` (and `util-core`
is declared only in `amber/build.sbt`). Folding those in means hosting
the util lower in the module
graph, tracked in #7095 rather than smuggled into this fix.
Log line, before and after:
```
- Failed to terminate region 1 on attempt 1 of 150. Retrying in 200 ms.
+ Failed to terminate region 1 on attempt 1 of 4. Retrying in 200 ms.
+ Failed to terminate region 1 on attempt 2 of 4. Retrying in 400 ms.
+ Failed to terminate region 1 on attempt 3 of 4. Retrying in 800 ms.
+ Region 1 could not be terminated after 4 attempts; giving up. Workers still not terminated: ...
```
This also shrinks -- but does not close -- the
retry-on-`JavaTimer`-thread race window in #6923, since there are far
fewer hops onto that thread.
### Any related issues, documentation, discussions?
Closes #7087
Related: #7056 (the 150-attempt line appears in those hang logs), #6891,
#6918, #6923. The bounded-retry loop this PR re-tunes was introduced in
#5737.
### How was this PR tested?
`UtilsSpec` covers `Utils.retry` directly: first-attempt success waits
not at all; the backoff
doubles (`200 -> 400 -> 800`); waiting stops as soon as an attempt
succeeds; `onRetry` sees the
1-based failed-attempt number and the upcoming backoff; a *synchronous*
throw from the by-name body
is retried like a failed `Future` rather than escaping; a *fatal* error
is not retried in either
shape (it escapes on attempt 1, spending no backoff); and `attempts <=
1` makes exactly one attempt
with no wait. The three cases that covered the removed blocking variant
are gone with it.
`RegionExecutionManagerSpec` keeps its termination-lifecycle cases and
pins what the defaults mean
end to end: the manager asks for `200 -> 400 -> 800 ms`, makes 4
`EndWorker` attempts, and a region
that terminates on attempt 2 spends only the first wait. Budget
boundaries are still pinned
(success on the last permitted attempt, give-up at 1 attempt / 2
attempts with every stuck worker
named).
The backoff assertions are exact rather than timing-tolerant:
`RecordingInlineTimer` records each
requested delay and runs the task inline, used inside
`Time.withCurrentTimeFrozen` so
`when - Time.now` is precisely what `Future.sleep` asked for. The tests
therefore assert the real
schedule without waiting 1.4 s or flaking on slow CI. That needed a
`killRetryTimer` seam on the
constructor (default `new JavaTimer(true)`, as before).
The two directly affected specs, with formatting and scalafix in the
same invocation:
```bash
sbt "WorkflowExecutionService/scalafmtAll" "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.common.UtilsSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec" "WorkflowExecutionService/scalafixAll --check"
```
```
[info] Tests: succeeded 27, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
Whole scheduling package plus `UtilsSpec`:
```bash
sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.* org.apache.texera.amber.engine.common.UtilsSpec"
```
```
[info] Suites: completed 13, aborted 1
[info] Tests: succeeded 145, failed 0, canceled 0, ignored 0, pending 0
```
(`DefaultCostEstimatorSpec` aborts on this machine during suite
construction, before any of its
tests run, while building a CSV scan descriptor. Pre-existing and
unrelated: with this PR's changes
reverted it aborts identically under the same command, 0 tests run.)
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---
.../scheduling/RegionExecutionManager.scala | 85 +++++++------
.../apache/texera/amber/engine/common/Utils.scala | 64 ++++++----
.../scheduling/RegionExecutionManagerSpec.scala | 89 ++++++++++---
.../RegionExecutionManagerTestSupport.scala | 6 +-
.../amber/engine/common/RecordingInlineTimer.scala | 51 ++++++++
.../texera/amber/engine/common/UtilsSpec.scala | 140 ++++++++++++++++++---
6 files changed, 337 insertions(+), 98 deletions(-)
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
index 3a87f8124e..0b32b74101 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
@@ -20,7 +20,7 @@
package org.apache.texera.amber.engine.architecture.scheduling
import org.apache.pekko.pattern.gracefulStop
-import com.twitter.util.{Duration => TwitterDuration, Future, JavaTimer, Return, Throw, Timer}
+import com.twitter.util.{Future, JavaTimer, Return, Throw, Timer}
import org.apache.texera.amber.core.state.State
import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory}
import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
@@ -51,7 +51,7 @@ import org.apache.texera.amber.engine.architecture.scheduling.config.{
ResourceConfig
}
import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning
-import org.apache.texera.amber.engine.common.AmberLogging
+import org.apache.texera.amber.engine.common.{AmberLogging, Utils}
import org.apache.texera.amber.engine.common.FutureBijection._
import org.apache.texera.amber.engine.common.rpc.AsyncRPCClient
import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR
@@ -64,11 +64,14 @@ import scala.concurrent.duration.{Duration => ScalaDuration}
object RegionExecutionManager {
- // Max EndWorker retries before termination fails. ~30s at DefaultKillRetryDelay (200ms).
- private[scheduling] val DefaultMaxTerminationAttempts: Int = 150
+ // Terminating a region is deterministic: `EndWorker` plus `gracefulStop` either succeed, or the
+ // engine has a bug that retrying cannot fix. Retries therefore only ride out a transient
+ // failure. With `Utils.retry`'s doubling backoff, 4 attempts from a 200ms base wait
+ // 200 + 400 + 800 ms = 1.4s in the worst case, where the former 150 attempts at a flat 200ms
+ // held a whole region's teardown for ~30s before the failure surfaced.
+ private[scheduling] val DefaultMaxTerminationAttempts: Int = 4
- private[scheduling] val DefaultKillRetryDelay: TwitterDuration =
- TwitterDuration.fromMilliseconds(200)
+ private[scheduling] val DefaultKillRetryBaseBackoffMs: Long = 200L
}
/**
@@ -105,8 +108,11 @@ class RegionExecutionManager(
coordinatorConfig: CoordinatorConfig,
actorService: PekkoActorService,
actorRefService: PekkoActorRefMappingService,
+ // Region-termination retry budget, handed to `Utils.retry`: `maxTerminationAttempts`
+ // attempts, waiting `killRetryBaseBackoffMs` before the first retry and doubling after each.
maxTerminationAttempts: Int = RegionExecutionManager.DefaultMaxTerminationAttempts,
- killRetryDelay: TwitterDuration = RegionExecutionManager.DefaultKillRetryDelay,
+ killRetryBaseBackoffMs: Long = RegionExecutionManager.DefaultKillRetryBaseBackoffMs,
+ killRetryTimer: Timer = new JavaTimer(true),
// Loop-back write addresses (Loop Start logical op id -> its input port's
// state URI), shipped to every worker in InitializeExecutorRequest. See
// WorkflowExecutionManager.loopStartStateUris.
@@ -125,7 +131,6 @@ class RegionExecutionManager(
Unexecuted
)
private val terminationFutureRef: AtomicReference[Future[Unit]] = new AtomicReference(null)
- private val killRetryTimer: Timer = new JavaTimer(true)
/**
* Sync the status of `RegionExecution` and transition this manager's phase to `Completed` only when the
@@ -231,38 +236,42 @@ class RegionExecutionManager(
}
}
- private def terminateWorkersWithRetry(
- regionExecution: RegionExecution,
- 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(", ")}).",
+ private def terminateWorkersWithRetry(regionExecution: RegionExecution): Future[Unit] = {
+ Utils
+ .retry(
+ attempts = maxTerminationAttempts,
+ baseBackoffTimeInMS = killRetryBaseBackoffMs,
+ timer = killRetryTimer,
+ onRetry = (err, attempt, backoffTimeInMS) =>
+ logger.warn(
+ s"Failed to terminate region ${region.id.id} on attempt $attempt of " +
+ s"$maxTerminationAttempts. Retrying in $backoffTimeInMS ms.",
err
)
- )
- case err =>
- logger.warn(
- s"Failed to terminate region ${region.id.id} on attempt $attempt of $maxTerminationAttempts. " +
- s"Retrying in ${killRetryDelay.inMilliseconds} ms.",
- err
- )
- Future
- .sleep(killRetryDelay)(killRetryTimer)
- .flatMap(_ => terminateWorkersWithRetry(regionExecution, attempt + 1))
- }
+ ) {
+ terminateWorkers(regionExecution)
+ }
+ .rescue {
+ // Every attempt failed. Name the workers that are still alive so the user can act on it.
+ case err =>
+ val workerIds = regionExecution.getAllOperatorExecutions.flatMap {
+ case (_, opExec) => opExec.getWorkerIds
+ }.toSeq
+ val attemptsLabel =
+ if (maxTerminationAttempts <= 1) "1 attempt" else s"$maxTerminationAttempts 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
+ )
+ )
+ }
}
def isCompleted: Boolean = currentPhaseRef.get == Completed
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
index 079640317c..5dee2c6754 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
@@ -19,12 +19,13 @@
package org.apache.texera.amber.engine.common
+import com.twitter.util.{Duration, Future, Timer}
import com.typesafe.scalalogging.LazyLogging
import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
import java.nio.file.{Files, Path, Paths}
import java.util.concurrent.locks.Lock
-import scala.annotation.tailrec
+import scala.util.control.NonFatal
object Utils extends LazyLogging {
@@ -60,32 +61,51 @@ object Utils extends LazyLogging {
val AMBER_HOME_FOLDER_NAME = "amber";
/**
- * Retry the given logic with a backoff time interval. The attempts are executed sequentially, thus blocking the thread.
- * Backoff time is doubled after each attempt.
+ * Retry the given logic with a backoff time interval, doubling the backoff after each attempt.
*
- * @param attempts total number of attempts. if n <= 1 then it will not retry at all, decreased by 1 for each recursion.
+ * The waits are scheduled on `timer`, so no thread is blocked between attempts. That matters
+ * for callers on an actor or coordinator thread, where a `Thread.sleep` would also stall
+ * unrelated work queued on that thread.
+ *
+ * A synchronous throw while evaluating `fn` counts as a failed attempt, same as a failed
+ * `Future`. Fatal errors are never retried in either shape: they propagate immediately.
+ *
+ * @param attempts total number of attempts. if n <= 1 then it will not retry at all.
* @param baseBackoffTimeInMS time to wait before next attempt, started with the base time, and doubled after each attempt.
- * @param fn the target function to execute.
- * @tparam T any return type from the provided function fn.
- * @return the provided function fn's return, or any exception that still being raised after n attempts.
+ * @param timer schedules the waits between attempts; no thread is blocked.
+ * @param onRetry invoked before each wait with the failure, the 1-based number of the
+ * attempt that just failed, and the backoff about to be waited in ms.
+ * Override it to log with caller context.
+ * @param fn the target function to execute, re-evaluated on each attempt.
+ * @tparam T any value type the provided function fn's `Future` yields.
+ * @return `fn`'s eventual value, or the last failure once `attempts` is exhausted.
*/
- @tailrec
- def retry[T](attempts: Int, baseBackoffTimeInMS: Long)(fn: => T): T = {
- try {
- fn
- } catch {
- case e: Throwable =>
- if (attempts > 1) {
- logger.warn(
- "retrying after " + baseBackoffTimeInMS + "ms, number of attempts left: " + (attempts - 1),
- e
- )
- Thread.sleep(baseBackoffTimeInMS)
- retry(attempts - 1, baseBackoffTimeInMS * 2)(fn)
- } else throw e
- }
+ def retry[T](
+ attempts: Int,
+ baseBackoffTimeInMS: Long,
+ timer: Timer,
+ onRetry: (Throwable, Int, Long) => Unit = logRetryAttempt
+ )(fn: => Future[T]): Future[T] = {
+ def attempt(attemptNumber: Int, backoffTimeInMS: Long): Future[T] =
+ Future(fn).flatten.rescue {
+ // `NonFatal` so that a fatal handed back as a failed `Future` is not retried either, which
+ // also matches how the blocking backoff loops elsewhere in the repo catch `Exception`.
+ case NonFatal(e) if attemptNumber < attempts =>
+ onRetry(e, attemptNumber, backoffTimeInMS)
+ Future
+ .sleep(Duration.fromMilliseconds(backoffTimeInMS))(timer)
+ .flatMap(_ => attempt(attemptNumber + 1, backoffTimeInMS * 2))
+ }
+
+ attempt(attemptNumber = 1, backoffTimeInMS = baseBackoffTimeInMS)
}
+ private def logRetryAttempt(failure: Throwable, attempt: Int, backoffTimeInMS: Long): Unit =
+ logger.warn(
+ "retrying after " + backoffTimeInMS + "ms, attempts made so far: " + attempt,
+ failure
+ )
+
private def isAmberHomePath(path: Path): Boolean = {
path.toRealPath().endsWith(AMBER_HOME_FOLDER_NAME)
}
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
index 62f59eee95..147a8e153a 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
@@ -19,7 +19,7 @@
package org.apache.texera.amber.engine.architecture.scheduling
-import com.twitter.util.{Duration => TwitterDuration, Future}
+import com.twitter.util.{Future, JavaTimer, Time, Timer}
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.testkit.TestKit
import org.apache.texera.amber.core.storage.VFSURIFactory
@@ -46,7 +46,7 @@ import org.apache.texera.amber.engine.architecture.scheduling.config.{
WorkerConfig
}
import org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState
-import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.{AmberRuntime, RecordingInlineTimer}
import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR
import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpecLike
@@ -77,6 +77,11 @@ class RegionExecutionManagerSpec
TestKit.shutdownActorSystem(system)
}
+ // Stand-in backoff for the tests that care about the retry budget (how many attempts it buys)
+ // rather than about the waits themselves. The real backoff values are asserted separately with
+ // `RecordingInlineTimer`.
+ private val fastRetryBackoffMs: Long = 5L
+
"RegionExecutionManager" should "send gracefulStop only after EndWorker succeeds" in {
val fixture = createSingleRegionFixture(endWorkerResponse = _ => None)
@@ -141,7 +146,7 @@ class RegionExecutionManagerSpec
val fixture = createSingleRegionFixture(
endWorkerResponse = _ => Some(transientEndWorkerFailure),
maxTerminationAttempts = 3,
- killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ killRetryBaseBackoffMs = fastRetryBackoffMs
)
launchRegion(fixture.manager)
@@ -157,10 +162,13 @@ class RegionExecutionManagerSpec
}
it should "give up after a single attempt when the budget is one" in {
+ // A budget of one means no retries: the first failure is final, and nothing is ever handed to
+ // the retry timer.
+ val timer = new RecordingInlineTimer
val fixture = createSingleRegionFixture(
endWorkerResponse = _ => Some(transientEndWorkerFailure),
maxTerminationAttempts = 1,
- killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ killRetryTimer = timer
)
launchRegion(fixture.manager)
@@ -172,6 +180,7 @@ class RegionExecutionManagerSpec
assert(failure.getMessage.contains("could not be terminated after 1 attempt"))
assert(failure.getMessage.contains(fixture.workerId.toString))
assert(fixture.rpcProbe.endWorkerCalls.size == 1)
+ assert(timer.recordedDelaysInMillis.isEmpty)
}
it should "complete when EndWorker finally succeeds on the last permitted attempt" in {
@@ -187,7 +196,7 @@ class RegionExecutionManagerSpec
Some(EmptyReturn())
},
maxTerminationAttempts = 2,
- killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ killRetryBaseBackoffMs = fastRetryBackoffMs
)
launchRegion(fixture.manager)
@@ -206,7 +215,7 @@ class RegionExecutionManagerSpec
val fixture = createMultiWorkerGiveUpFixture(
workerCount = 3,
maxTerminationAttempts = 2,
- killRetryDelay = TwitterDuration.fromMilliseconds(5)
+ killRetryBaseBackoffMs = fastRetryBackoffMs
)
launchRegion(fixture.manager)
@@ -225,13 +234,57 @@ class RegionExecutionManagerSpec
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(RegionExecutionManager.DefaultMaxTerminationAttempts == 150)
- assert(
- RegionExecutionManager.DefaultKillRetryDelay == TwitterDuration.fromMilliseconds(200)
- )
+ 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.
+ assert(RegionExecutionManager.DefaultMaxTerminationAttempts == 4)
+ assert(RegionExecutionManager.DefaultKillRetryBaseBackoffMs == 200L)
+ }
+
+ it should "back off exponentially and stop once the retry budget is exhausted" in {
+ // Pins what the defaults mean in practice: the manager asks for 200, then 400, then 800 ms,
+ // and makes 4 EndWorker attempts. Asserted through the timer rather than through
+ // `Utils.retry`'s own unit tests, so a wrong base or attempt count here is caught too.
+ // The inline timer runs each wait immediately, so this costs no wall-clock time.
+ val timer = new RecordingInlineTimer
+ Time.withCurrentTimeFrozen { _ =>
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => Some(transientEndWorkerFailure),
+ killRetryTimer = timer
+ )
+
+ launchRegion(fixture.manager)
+ val failure = intercept[IllegalStateException] {
+ await(requestRegionCompletion(fixture.manager))
+ }
+
+ assert(failure.getMessage.contains("could not be terminated after 4 attempts"))
+ assert(fixture.rpcProbe.endWorkerCalls.size == 4)
+ assert(timer.recordedDelaysInMillis == Seq(200L, 400L, 800L))
+ }
+ }
+
+ it should "spend only the delays it needs when a retry succeeds" in {
+ // A region that terminates on attempt 2 must wait 200 ms once and stop there, rather than
+ // walking the rest of the backoff.
+ val attempts = new atomic.AtomicInteger(0)
+ val timer = new RecordingInlineTimer
+ Time.withCurrentTimeFrozen { _ =>
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ =>
+ if (attempts.incrementAndGet() == 1) Some(transientEndWorkerFailure)
+ else Some(EmptyReturn()),
+ killRetryTimer = timer
+ )
+
+ launchRegion(fixture.manager)
+ await(requestRegionCompletion(fixture.manager))
+
+ assert(fixture.manager.isCompleted)
+ assert(fixture.rpcProbe.endWorkerCalls.size == 2)
+ assert(timer.recordedDelaysInMillis == Seq(200L))
+ }
}
it should "surface the underlying cause when an output port schema is unavailable" in {
@@ -326,7 +379,8 @@ class RegionExecutionManagerSpec
private def createSingleRegionFixture(
endWorkerResponse: WorkerRpcCall => Option[ControlReturn],
maxTerminationAttempts: Int = RegionExecutionManager.DefaultMaxTerminationAttempts,
- killRetryDelay: TwitterDuration = RegionExecutionManager.DefaultKillRetryDelay
+ killRetryBaseBackoffMs: Long = RegionExecutionManager.DefaultKillRetryBaseBackoffMs,
+ killRetryTimer: Timer = new JavaTimer(true)
): SingleRegionFixture = {
val physicalOp = createSourceOp("test-op")
val workerId = createWorkerId(physicalOp)
@@ -355,7 +409,8 @@ class RegionExecutionManagerSpec
coordinator.actorService,
coordinator.actorRefService,
maxTerminationAttempts,
- killRetryDelay
+ killRetryBaseBackoffMs,
+ killRetryTimer
)
SingleRegionFixture(
@@ -380,7 +435,7 @@ class RegionExecutionManagerSpec
private def createMultiWorkerGiveUpFixture(
workerCount: Int,
maxTerminationAttempts: Int,
- killRetryDelay: TwitterDuration
+ killRetryBaseBackoffMs: Long
): MultiWorkerFixture = {
val physicalOp = createSourceOp("multi-op")
val workerIds = createWorkerIds(physicalOp, workerCount)
@@ -403,7 +458,7 @@ class RegionExecutionManagerSpec
coordinator.actorService,
coordinator.actorRefService,
maxTerminationAttempts,
- killRetryDelay
+ killRetryBaseBackoffMs
)
MultiWorkerFixture(manager, rpcProbe, workerIds)
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala
index 5fc6375ddd..6840eaf9b7 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala
@@ -66,9 +66,9 @@ object RegionExecutionManagerTestSupport {
val StartWorker = "startWorker"
val EndWorker = "endWorker"
- // Generous deadline for the polling helpers below. Production timing under test (notably the
- // 200 ms `killRetryDelay` in `RegionExecutionManager`) fits comfortably; the rest is
- // headroom for slow CI.
+ // Generous deadline for the polling helpers below. Production timing under test (notably
+ // `RegionExecutionManager`'s default 200/400/800 ms termination backoff) fits comfortably; the
+ // rest is headroom for slow CI.
val testTimeout: Duration = Duration.fromSeconds(5)
case class WorkerRpcCall(
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/common/RecordingInlineTimer.scala b/amber/src/test/scala/org/apache/texera/amber/engine/common/RecordingInlineTimer.scala
new file mode 100644
index 0000000000..7efe791bbf
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/common/RecordingInlineTimer.scala
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.common
+
+import com.twitter.util.{Duration, Time, Timer, TimerTask}
+
+import scala.collection.mutable
+
+/**
+ * A `Timer` that records the delay of every task scheduled on it and then runs the task inline
+ * instead of waiting.
+ *
+ * Use it inside `Time.withCurrentTimeFrozen` so that `when - Time.now` is exactly the delay
+ * `Future.sleep` asked for. That makes a backoff schedule (e.g. `Utils.retry`'s) assertable
+ * exactly, on the test thread, without the test spending that time asleep.
+ */
+class RecordingInlineTimer extends Timer {
+ private val delays: mutable.ArrayBuffer[Duration] = mutable.ArrayBuffer()
+
+ def recordedDelays: Seq[Duration] = delays.toSeq
+
+ def recordedDelaysInMillis: Seq[Long] = delays.map(_.inMilliseconds).toSeq
+
+ def scheduleOnce(when: Time)(f: => Unit): TimerTask = {
+ delays += (when - Time.now)
+ f
+ new TimerTask { def cancel(): Unit = () }
+ }
+
+ def schedulePeriodically(when: Time, period: Duration)(f: => Unit): TimerTask =
+ throw new AssertionError("retry backoff must not schedule periodic tasks")
+
+ def stop(): Unit = ()
+}
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala
index 13674139f1..640a691e78 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala
@@ -19,10 +19,12 @@
package org.apache.texera.amber.engine.common
+import com.twitter.util.{Await, Future, Time}
import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
import org.scalatest.flatspec.AnyFlatSpec
import java.util.concurrent.locks.ReentrantLock
+import scala.collection.mutable
class UtilsSpec extends AnyFlatSpec {
@@ -99,37 +101,139 @@ class UtilsSpec extends AnyFlatSpec {
// -- retry ---------------------------------------------------------------
- "Utils.retry" should "return the value on the first successful attempt without retrying" in {
+ // `retry` is exercised with `RecordingInlineTimer`: it captures the delay of each scheduled wait
+ // and runs it immediately, so the backoff schedule is asserted exactly and no test spends that
+ // time asleep. `Time.withCurrentTimeFrozen` makes `when - Time.now` equal the delay
+ // `Future.sleep` asked for.
+
+ "Utils.retry" should "return the value on the first successful attempt without waiting" in {
+ val timer = new RecordingInlineTimer
var calls = 0
- val result = Utils.retry(attempts = 3, baseBackoffTimeInMS = 0L) {
- calls += 1
- "ok"
- }
+ val result = Await.result(
+ Utils.retry(attempts = 3, baseBackoffTimeInMS = 200L, timer = timer) {
+ calls += 1
+ Future.value("ok")
+ }
+ )
assert(result == "ok")
assert(calls == 1)
+ assert(timer.recordedDelaysInMillis.isEmpty)
}
- it should "retry on failure until success and return the eventual result" in {
+ it should "double the backoff after each failed attempt" in {
+ val timer = new RecordingInlineTimer
var calls = 0
- val result = Utils.retry(attempts = 3, baseBackoffTimeInMS = 0L) {
- calls += 1
- if (calls < 2) throw new RuntimeException("transient")
- "ok"
+ Time.withCurrentTimeFrozen { _ =>
+ val failure = intercept[RuntimeException] {
+ Await.result(
+ Utils.retry(attempts = 4, baseBackoffTimeInMS = 200L, timer = timer) {
+ calls += 1
+ Future.exception(new RuntimeException(s"failure-$calls"))
+ }
+ )
+ }
+ // The last failure is what surfaces, and 4 attempts spend exactly 3 waits.
+ assert(failure.getMessage == "failure-4")
+ assert(calls == 4)
+ assert(timer.recordedDelaysInMillis == Seq(200L, 400L, 800L))
}
- assert(result == "ok")
- assert(calls == 2)
}
- it should "rethrow the last exception after exhausting all attempts" in {
+ it should "stop waiting as soon as an attempt succeeds" in {
+ val timer = new RecordingInlineTimer
var calls = 0
- val ex = intercept[RuntimeException] {
- Utils.retry(attempts = 2, baseBackoffTimeInMS = 0L) {
- calls += 1
- throw new RuntimeException(s"failure-$calls")
+ Time.withCurrentTimeFrozen { _ =>
+ val result = Await.result(
+ Utils.retry(attempts = 4, baseBackoffTimeInMS = 200L, timer = timer) {
+ calls += 1
+ if (calls < 3) Future.exception(new RuntimeException("transient"))
+ else Future.value(calls)
+ }
+ )
+ assert(result == 3)
+ assert(timer.recordedDelaysInMillis == Seq(200L, 400L))
+ }
+ }
+
+ it should "report the failed attempt number and the upcoming backoff to onRetry" in {
+ val timer = new RecordingInlineTimer
+ val observed = mutable.ArrayBuffer[(String, Int, Long)]()
+ Time.withCurrentTimeFrozen { _ =>
+ intercept[RuntimeException] {
+ Await.result(
+ Utils.retry(
+ attempts = 3,
+ baseBackoffTimeInMS = 200L,
+ timer = timer,
+ onRetry =
+ (err, attempt, backoffMs) => observed += ((err.getMessage, attempt, backoffMs))
+ ) {
+ Future.exception(new RuntimeException("boom"))
+ }
+ )
}
+ // Called once per wait (never after the final attempt), with 1-based attempt numbers.
+ assert(observed.toSeq == Seq(("boom", 1, 200L), ("boom", 2, 400L)))
}
+ }
+
+ it should "retry a synchronous throw from the body, not just a failed Future" in {
+ // The body is by-name, so a `Future`-returning expression can still blow up before it ever
+ // produces a Future; that must be retried like any other failure rather than escaping.
+ val timer = new RecordingInlineTimer
+ var calls = 0
+ val result = Await.result(
+ Utils.retry(attempts = 3, baseBackoffTimeInMS = 1L, timer = timer) {
+ calls += 1
+ if (calls < 2) throw new IllegalStateException("sync boom")
+ Future.value("ok")
+ }
+ )
+ assert(result == "ok")
assert(calls == 2)
- assert(ex.getMessage == "failure-2")
+ }
+
+ it should "not retry a fatal error, whether thrown or returned as a failed Future" in {
+ // A fatal is not a transient failure, so it must escape on the first attempt. The two shapes
+ // travel different paths -- a synchronous throw is filtered by `Future(fn)`'s `Try`, a failed
+ // `Future` reaches the `rescue` guard -- and both have to behave the same way.
+ Seq[(String, () => Future[String])](
+ "as a failed Future" -> (() => Future.exception(new InterruptedException("fatal"))),
+ "thrown synchronously" -> (() => throw new InterruptedException("fatal"))
+ ).foreach {
+ case (shape, body) =>
+ val timer = new RecordingInlineTimer
+ var calls = 0
+ intercept[InterruptedException] {
+ Await.result(
+ Utils.retry(attempts = 4, baseBackoffTimeInMS = 200L, timer = timer) {
+ calls += 1
+ body()
+ }
+ )
+ }
+ assert(calls == 1, s"fatal $shape was retried")
+ assert(timer.recordedDelaysInMillis.isEmpty, s"fatal $shape caused a backoff wait")
+ }
+ }
+
+ it should "make a single attempt when attempts is one or less" in {
+ // A budget of 1 -- or a nonsensical 0 -- means no retry at all, and no wait either.
+ Seq(1, 0).foreach { attempts =>
+ val timer = new RecordingInlineTimer
+ var calls = 0
+ val failure = intercept[RuntimeException] {
+ Await.result(
+ Utils.retry(attempts = attempts, baseBackoffTimeInMS = 200L, timer = timer) {
+ calls += 1
+ Future.exception(new RuntimeException("only-once"))
+ }
+ )
+ }
+ assert(failure.getMessage == "only-once", s"attempts = $attempts")
+ assert(calls == 1, s"attempts = $attempts")
+ assert(timer.recordedDelaysInMillis.isEmpty, s"attempts = $attempts")
+ }
}
// -- withLock ------------------------------------------------------------