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 <[email protected]> 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 ------------------------------------------------------------
