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-7119-ac68ea471bc8367d6142ddc9f4ae7df6f1aa3d06 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2b030973f3a2ab8356307e36a5cd2a5998b1b99d Author: Xinyuan Lin <[email protected]> AuthorDate: Thu Jul 30 15:14:01 2026 -0700 refactor(util): host one exponential-backoff retry (#7119) ### What changes were proposed in this PR? The same doubling-backoff loop was written three times, and no module could share the others' copy: amber's `Utils` sits above `common/workflow-core` and `file-service` in the module graph. New dependency-free **`common/util`** module with `RetryUtil.withBackoff`, and both blocking copies now call it: | Site | Before | After | | --- | --- | --- | | `LakeFSStorageClient.healthCheck` | private 27-line `retryWithBackoff` | `RetryUtil.withBackoff("connect to lake fs server", 5, 200, ...)` | | `FileService.awaitDependency` | private 36-line loop | 7-line delegation keeping its 6/200 defaults + logger | | `Utils.retry` (amber, non-blocking) | unchanged | cross-references the blocking sibling | ```scala RetryUtil.withBackoff(description, maxAttempts, initialDelayMillis, onRetry, sleep)(operation) ``` The util owns everything the copies duplicated: attempt counting, the doubling, what counts as transient (`NonFatal`), interrupt fail-fast with the interrupt status restored, and the message wording. `description` is a verb phrase ("connect to lake fs server"), interpolated into all three messages, so LakeFS's give-up text is byte-identical to before. Retries are reported through an `onRetry` hook carrying a `RetryAttempt` (attempt, budget, delay, cause, and the standard `message`), which callers log with **their own** logger — so a LakeFS retry still logs under LakeFS, not under the util. **Bug fixed on the way in.** LakeFS's `sleep` call sat inside the `catch` block, so an interrupt raised *while waiting* escaped raw with the interrupt flag left cleared — a `catch` cannot catch what its own body throws: ``` Before (LakeFS): op fails -> catch -> sleep interrupted -> InterruptedException escapes, flag cleared After (shared): op fails -> sleep interrupted -> interrupt restored + wrapped, fails fast ``` `FileService`'s copy already plugged that hole with an inner `try`; the shared util keeps the better behavior for both. **Why a new module rather than `common/workflow-core`**, which was proposal 1 in #7095: `workflow-core` is not reachable from `Auth`, `ConfigService`, or `AccessControlService`, so hosting it there would have left future callers in those modules stuck writing their own loop — the exact thing this PR is meant to stop. `common/config` is reachable from everything and already carries `org.apache.texera.amber.util.ConfigParserUtil`, but a retry helper isn't configuration. A dependency-free module any other module may depend on avoids both problems, at the cost of one `build.sbt` entry. **Why the non-blocking variant stays in amber.** `Utils.retry` needs `com.twitter:util-core`, declared only in `amber/build.sbt`. Moving it down would put util-core on the classpath of every service that depends on the new module and force LICENSE-binary resyncs, so the pair is: blocking in `common/util`, non-blocking in `amber`, each documented in terms of the other. The new module is dependency-free for the same reason. **One behavior change, deliberate.** Both replaced loops caught `case e: Exception`; the util uses `NonFatal`, for parity with `Utils.retry`. `NonFatal` is wider: non-fatal `Error`s — `AssertionError`, `java.io.IOError`, `ServiceConfigurationError` — are now retried and wrapped rather than propagating immediately. On the S3/LakeFS startup paths that means such a failure costs the full backoff before surfacing. `RetryUtilSpec` pins it in both directions (a non-fatal `Error` is retried; a fatal throwable is not). **Two retry sites deliberately left alone**, because converting them changes behavior rather than just structure — tracked with the full reasoning in #7124: | Site | Why not now | | --- | --- | | `URLFetchUtil.getInputStreamFromURL` | Retries 5x with **no** delay and returns `Option`. Adopting backoff adds up to ~3 s before a dead URL gives up — arguably a fix (it is an unthrottled retry storm today), but a behavior change in an operator path. | | `PythonProxyClient` Flight connect loop | **Constant** delay, closes the Flight client between attempts, and throws `WorkflowRuntimeException` on give-up. Needs a delay-multiplier knob plus a give-up type decision. | ### Any related issues, documentation, discussions? Closes #7095 Follows #7088, which introduced the non-blocking variant; the consolidation was split out of it to keep that fix narrow. Context: the review discussion on #7103 (the LakeFS health-check backport) asked for one util rather than a fourth copy. ### How was this PR tested? `RetryUtilSpec` (new, in `common/util`, registered in `build.yml`'s hand-maintained `jacoco` module list so it actually runs) is the union of what the two hand-rolled loops were tested for, plus what neither covered: - returns the operation's value on first success without sleeping; retries until success with a doubling progression; honors a custom base delay; succeeds on the final permitted attempt (the `attempt >= maxAttempts` boundary); gives up naming the description and preserving the cause; `maxAttempts = 1` means no retry and no sleep; - the `onRetry` hook fires once per wait with 1-based attempt numbers and never after the last attempt, and its `message` wording is pinned; - interrupt during the operation **and** interrupt during a backoff sleep both restore the interrupt status and fail fast (the second is the LakeFS hole above); - a fatal throwable is not retried and passes through unwrapped, while a non-fatal `Error` is retried (the `NonFatal`-vs-`Exception` widening above). At the call sites: `FileServiceSpec` keeps 8 tests that exercise the delegation, its own 6/200 defaults, and the description in its messages (the 4 that only re-tested util mechanics moved into `RetryUtilSpec`). `LakeFSStorageClientSpec`'s 4 retry tests moved out with the loop; its remaining 5 pass unchanged. amber's `UtilsSpec` and `RegionExecutionManagerSpec` are untouched and still green. ```bash sbt "Util/jacoco" "FileService/testOnly org.apache.texera.service.FileServiceSpec" "WorkflowCore/testOnly org.apache.texera.amber.core.storage.util.LakeFSStorageClientSpec" ``` ``` [info] Tests: succeeded 11, failed 0, canceled 0, ignored 0, pending 0 (Util) [info] Tests: succeeded 12, failed 0, canceled 0, ignored 0, pending 0 (FileService) [info] Tests: succeeded 5, failed 0, canceled 0, ignored 0, pending 0 (WorkflowCore / LakeFS) ``` Lint plus test-source compiles for every module the new dependency edge touches, and amber's two retry specs: ```bash sbt "Util/scalafixAll --check" "Util/scalafmtCheckAll" "WorkflowCore/Test/compile" "WorkflowCore/scalafixAll --check" "FileService/Test/compile" "FileService/scalafixAll --check" "WorkflowExecutionService/Test/compile" "WorkflowExecutionService/scalafixAll --check" "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.common.UtilsSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec" ``` ``` [success] all scalafix / scalafmt checks and Test/compile tasks [info] Tests: succeeded 27, failed 0, canceled 0, ignored 0, pending 0 ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .github/workflows/build.yml | 1 + AGENTS.md | 2 +- .../apache/texera/amber/engine/common/Utils.scala | 9 +- build.sbt | 8 +- common/util/build.sbt | 56 ++++++ .../org/apache/texera/common/util/RetryUtil.scala | 118 ++++++++++++ .../apache/texera/common/util/RetryUtilSpec.scala | 206 +++++++++++++++++++++ .../core/storage/util/LakeFSStorageClient.scala | 50 +---- .../storage/util/LakeFSStorageClientSpec.scala | 49 +---- .../org/apache/texera/service/FileService.scala | 57 ++---- .../apache/texera/service/FileServiceSpec.scala | 89 ++------- 11 files changed, 433 insertions(+), 212 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index becd8deb34..a82cbfde95 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -309,6 +309,7 @@ jobs: "Auth/jacoco" \ "Config/jacoco" \ "Resource/jacoco" \ + "Util/jacoco" \ "PyBuilder/jacoco" \ "WorkflowCore/jacoco" \ "WorkflowOperator/jacoco" \ diff --git a/AGENTS.md b/AGENTS.md index b21ed6d6a1..7118073a78 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,7 +29,7 @@ engine, an Angular UI, and the agent service. JVM modules wired in | --- | --- | --- | | Workflow execution engine (Amber) | `amber/` | [amber/README.md](amber/README.md) | | Backend services | `config-service/`, `access-control-service/`, `file-service/`, `computing-unit-managing-service/`, `workflow-compiling-service/`, `notebook-migration-service/` | `build.sbt` | -| Shared Scala libs | `common/` (`auth`, `config`, `dao`, `workflow-core`, `workflow-operator`, `pybuilder`) | `build.sbt` | +| Shared Scala libs | `common/` (`auth`, `config`, `dao`, `util`, `workflow-core`, `workflow-operator`, `pybuilder`) | `build.sbt` | | Frontend (Angular) | `frontend/` | [frontend/README.md](frontend/README.md) | | Agent service (Bun/TS, LLM agents) | `agent-service/` | `agent-service/package.json` | | Pyright language service | `pyright-language-service/` | [pyright-language-service/README.md](pyright-language-service/README.md) | 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 5dee2c6754..393e5b343a 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 @@ -67,6 +67,11 @@ object Utils extends LazyLogging { * for callers on an actor or coordinator thread, where a `Thread.sleep` would also stall * unrelated work queued on that thread. * + * This is the non-blocking half of the pair: for blocking work, use + * `org.apache.texera.common.util.RetryUtil.withBackoff` in `common/util`, which takes the same + * attempts-and-doubling knobs. It lives in a separate module because this variant needs + * `com.twitter:util-core`, which only `amber` declares. + * * 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. * @@ -88,8 +93,8 @@ object Utils extends LazyLogging { )(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`. + // `NonFatal` so that a fatal handed back as a failed `Future` is not retried either, matching + // the blocking sibling `RetryUtil.withBackoff`, which uses the same `NonFatal` predicate. case NonFatal(e) if attemptNumber < attempts => onRetry(e, attemptNumber, backoffTimeInMS) Future diff --git a/build.sbt b/build.sbt index 720fc28b80..b10661fd70 100644 --- a/build.sbt +++ b/build.sbt @@ -117,6 +117,9 @@ val nettyDependencyOverrides = Seq( // keep the org.apache.log4j API available at runtime. ThisBuild / excludeDependencies += ExclusionRule("log4j", "log4j") +// Dependency-free helpers (retry/backoff, ...) that any module may depend on. Keep it that way: +// anything added here reaches the classpath of every service that depends on it. +lazy val Util = (project in file("common/util")).settings(commonModuleSettings) lazy val DAO = (project in file("common/dao")).settings(commonModuleSettings) lazy val Config = (project in file("common/config")).settings(commonModuleSettings) lazy val Resource = (project in file("common/resource")).settings(commonModuleSettings) @@ -156,7 +159,7 @@ lazy val PyBuilder = (project in file("common/pybuilder")) lazy val WorkflowCore = (project in file("common/workflow-core")) .settings(commonModuleSettings) - .dependsOn(DAO, Config, PyBuilder) + .dependsOn(DAO, Config, PyBuilder, Util) .configs(Test) .dependsOn(DAO % "test->test") // test scope dependency lazy val ComputingUnitManagingService = (project in file("computing-unit-managing-service")) @@ -202,7 +205,7 @@ lazy val ComputingUnitManagingService = (project in file("computing-unit-managin ) lazy val FileService = (project in file("file-service")) .settings(commonModuleSettings) - .dependsOn(WorkflowCore, Auth, Config, Resource) + .dependsOn(WorkflowCore, Auth, Config, Resource, Util) .configs(Test) .dependsOn(DAO % "test->test") // test scope dependency .settings( @@ -278,6 +281,7 @@ lazy val TexeraProject = (project in file(".")) Auth, Config, Resource, + Util, DAO, PyBuilder, WorkflowCore, diff --git a/common/util/build.sbt b/common/util/build.sbt new file mode 100644 index 0000000000..0f4446edcc --- /dev/null +++ b/common/util/build.sbt @@ -0,0 +1,56 @@ +// 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. + +import scala.collection.Seq + +name := "util" + +enablePlugins(JavaAppPackaging) + +// Enable semanticdb for Scalafix +ThisBuild / semanticdbEnabled := true +ThisBuild / semanticdbVersion := scalafixSemanticdb.revision + +// Manage dependency conflicts by always using the latest revision +ThisBuild / conflictManager := ConflictManager.latestRevision + +// Restrict parallel execution of tests to avoid conflicts +Global / concurrentRestrictions += Tags.limit(Tags.Test, 1) + +///////////////////////////////////////////////////////////////////////////// +// Compiler Options +///////////////////////////////////////////////////////////////////////////// + +// Scala compiler options +Compile / scalacOptions ++= Seq( + "-Xelide-below", "WARNING", // Turn on optimizations with "WARNING" as the threshold + "-feature", // Check feature warnings + "-deprecation", // Check deprecation warnings + "-Ywarn-unused:imports" // Check for unused imports +) + +///////////////////////////////////////////////////////////////////////////// +// Dependencies +///////////////////////////////////////////////////////////////////////////// + +// This module is deliberately dependency-free apart from the test framework: +// any module may depend on it, so anything added here reaches the classpath of +// every service that does. Callers pass their own logger in through the hooks +// instead of the module pulling in a logging library. +libraryDependencies ++= Seq( + "org.scalatest" %% "scalatest" % "3.2.15" % Test // ScalaTest (for unit tests) +) diff --git a/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala new file mode 100644 index 0000000000..3ebf094b4e --- /dev/null +++ b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala @@ -0,0 +1,118 @@ +/* + * 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.common.util + +import scala.annotation.tailrec +import scala.util.control.NonFatal + +/** + * Retry with exponential backoff, for blocking work. + * + * This module has no dependencies of its own, so any module may depend on it rather than writing + * another retry loop by hand. If the work returns a `Future` rather than blocking, use the + * non-blocking sibling + * `org.apache.texera.amber.engine.common.Utils.retry` in `amber` instead: it takes the same + * attempts-and-doubling-backoff knobs but waits on a `Timer`, which matters on an actor or + * coordinator thread where `Thread.sleep` would stall unrelated work queued behind it. + */ +object RetryUtil { + + /** + * One failed attempt that is about to be retried. Carries everything a caller needs to log the + * retry itself; `message` is the standard wording, so retries read the same everywhere. + */ + final case class RetryAttempt( + description: String, + attempt: Int, + maxAttempts: Int, + delayMillis: Long, + cause: Throwable + ) { + def message: String = + s"Failed to $description (attempt $attempt/$maxAttempts): ${cause.getMessage}. " + + s"Retrying in ${delayMillis}ms..." + } + + /** + * Runs `operation`, retrying on failure with exponential backoff (the delay doubles after each + * failed attempt) until it succeeds or `maxAttempts` is reached. The final failure is wrapped + * with `description` and the last exception as its cause. + * + * Only `NonFatal` failures are treated as transient, which is the same predicate the + * non-blocking sibling uses. Note that `NonFatal` admits non-fatal `Error`s -- `AssertionError`, + * `java.io.IOError`, `ServiceConfigurationError` -- so those are retried rather than propagated + * straight away. An `InterruptedException` -- raised by the operation or by the wait between + * attempts -- fails fast with the interrupt status restored, so a caller shutting the thread + * down is never made to sit through the remaining backoff. + * + * @param description verb phrase naming the work, e.g. "connect to lake fs server". It is + * interpolated into every message: "Failed to $description after ...". + * @param maxAttempts total attempts; 1 means no retry at all. + * @param initialDelayMillis wait before the first retry; doubled after each failed attempt. + * @param onRetry invoked before each wait. Log `RetryAttempt.message` through the + * caller's own logger, so retries are attributed to the caller rather + * than to this util. + * @param sleep how to wait; injectable so tests exercise the backoff without waiting. + * @param operation the work to run, re-evaluated on each attempt. + * @tparam T whatever `operation` returns. + * @return `operation`'s value from the first attempt that succeeds. + */ + def withBackoff[T]( + description: String, + maxAttempts: Int, + initialDelayMillis: Long, + onRetry: RetryAttempt => Unit, + sleep: Long => Unit = Thread.sleep + )(operation: => T): T = { + // Restore the interrupt status and fail fast rather than retrying, whether the interrupt + // arrives while running `operation` or while waiting between attempts. + def failInterrupted(cause: InterruptedException): Nothing = { + Thread.currentThread().interrupt() + throw new RuntimeException(s"Interrupted while waiting to $description", cause) + } + + @tailrec + def attemptFrom(attempt: Int, delayMillis: Long): T = { + val outcome: Either[Throwable, T] = + try Right(operation) + catch { + case ie: InterruptedException => failInterrupted(ie) + case NonFatal(cause) => Left(cause) + } + + outcome match { + case Right(value) => value + case Left(cause) => + if (attempt >= maxAttempts) { + throw new RuntimeException( + s"Failed to $description after $maxAttempts attempts: ${cause.getMessage}", + cause + ) + } + onRetry(RetryAttempt(description, attempt, maxAttempts, delayMillis, cause)) + try sleep(delayMillis) + catch { case ie: InterruptedException => failInterrupted(ie) } + attemptFrom(attempt + 1, delayMillis * 2) + } + } + + attemptFrom(attempt = 1, delayMillis = initialDelayMillis) + } +} diff --git a/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala new file mode 100644 index 0000000000..828cb8d01e --- /dev/null +++ b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala @@ -0,0 +1,206 @@ +/* + * 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.common.util + +import org.apache.texera.common.util.RetryUtil.RetryAttempt +import org.scalatest.flatspec.AnyFlatSpec + +import scala.collection.mutable.ListBuffer +import scala.util.control.ControlThrowable + +/** + * Contract of the shared blocking backoff retry. `sleep` is injected everywhere so the backoff + * progression is asserted exactly without any test waiting. + * + * Coverage is the full contract both blocking callers rely on: the doubling progression, which + * failures count as transient, the give-up wrapping, and interrupt fail-fast during the operation + * and during a backoff sleep. + */ +class RetryUtilSpec extends AnyFlatSpec { + + private def noopRetryHook: RetryAttempt => Unit = _ => () + + "RetryUtil.withBackoff" should "return the operation's value on first success without sleeping" in { + val delays = ListBuffer.empty[Long] + var attempts = 0 + val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { + attempts += 1 + "value" + } + assert(result == "value") + assert(attempts == 1) + assert(delays.isEmpty) + } + + it should "retry until success and double the delay after each failed attempt" in { + val delays = ListBuffer.empty[Long] + var attempts = 0 + val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { + attempts += 1 + if (attempts < 3) throw new RuntimeException("transient") + attempts + } + assert(result == 3) + assert(delays.toList == List(200L, 400L)) + } + + it should "honor a custom initial delay when computing the progression" in { + // Guards against a hardcoded base: from 50ms the progression must be 50, 100, 200. + val delays = ListBuffer.empty[Long] + intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 4, 50L, noopRetryHook, delays += _) { + throw new RuntimeException("down") + } + } + assert(delays.toList == List(50L, 100L, 200L)) + } + + it should "succeed on the final permitted attempt without giving up one try too early" in { + // Boundary for `attempt >= maxAttempts`: success on the very last attempt must still count. + val delays = ListBuffer.empty[Long] + var attempts = 0 + RetryUtil.withBackoff("reach the store", 3, 200L, noopRetryHook, delays += _) { + attempts += 1 + if (attempts < 3) throw new RuntimeException("transient") + } + assert(attempts == 3) + assert(delays.toList == List(200L, 400L)) + } + + it should "give up after maxAttempts, naming the description and preserving the cause" in { + val cause = new RuntimeException("still down") + var attempts = 0 + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff("connect to lake fs server", 3, 200L, noopRetryHook, _ => ()) { + attempts += 1 + throw cause + } + } + assert(attempts == 3) + assert(failure.getMessage == "Failed to connect to lake fs server after 3 attempts: still down") + assert(failure.getCause eq cause) + } + + it should "give up immediately without sleeping when maxAttempts is one" in { + val delays = ListBuffer.empty[Long] + var attempts = 0 + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 1, 200L, noopRetryHook, delays += _) { + attempts += 1 + throw new RuntimeException("still down") + } + } + assert(attempts == 1) + assert(delays.isEmpty) + assert(failure.getMessage.contains("after 1 attempts")) + } + + it should "report each retry to the hook, and never after the last attempt" in { + val observed = ListBuffer.empty[RetryAttempt] + intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 3, 200L, observed += _, _ => ()) { + throw new RuntimeException("down") + } + } + // 3 attempts buy 2 retries, so the hook fires twice with 1-based attempt numbers. + assert( + observed.map(a => (a.attempt, a.maxAttempts, a.delayMillis)).toList == List( + (1, 3, 200L), + (2, 3, 400L) + ) + ) + assert(observed.forall(_.cause.getMessage == "down")) + assert( + observed.head.message == + "Failed to reach the store (attempt 1/3): down. Retrying in 200ms..." + ) + } + + it should "fail fast and restore the interrupt status when the operation is interrupted" in { + val delays = ListBuffer.empty[Long] + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { + throw new InterruptedException("interrupted") + } + } + // Thread.interrupted() both reads and clears the flag, so the interrupt was restored. + assert(Thread.interrupted()) + assert(failure.getMessage == "Interrupted while waiting to reach the store") + assert(failure.getCause.isInstanceOf[InterruptedException]) + assert(delays.isEmpty) + } + + it should "fail fast and restore the interrupt status when interrupted during a backoff sleep" in { + // A `catch` cannot catch what its own body throws, so an interrupt raised by the wait needs + // handling of its own; without it the exception escapes raw and the flag stays cleared. + var attempts = 0 + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff( + "reach the store", + 5, + 200L, + noopRetryHook, + _ => throw new InterruptedException("interrupted") + ) { + attempts += 1 + throw new RuntimeException("transient") + } + } + assert(attempts == 1) + assert(Thread.interrupted()) + assert(failure.getMessage == "Interrupted while waiting to reach the store") + assert(failure.getCause.isInstanceOf[InterruptedException]) + } + + it should "not retry a fatal throwable, and let it through unwrapped" in { + // Only `NonFatal` failures are transient. A control throwable stands in for a fatal here + // because a real `VirtualMachineError` would abort the suite rather than be caught. + val delays = ListBuffer.empty[Long] + var attempts = 0 + object Fatal extends ControlThrowable + intercept[ControlThrowable] { + RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { + attempts += 1 + throw Fatal + } + } + assert(attempts == 1) + assert(delays.isEmpty) + } + + it should "retry a non-fatal Error, which the predicate admits despite it not being an Exception" in { + // `NonFatal` is wider than `Exception`: an `AssertionError` (or `java.io.IOError`, + // `ServiceConfigurationError`) is transient here, so it is retried and then wrapped like any + // other failure. Pinned because it is the one behavior difference from the `case e: Exception` + // loops this util replaced. + val delays = ListBuffer.empty[Long] + var attempts = 0 + val cause = new AssertionError("assertion blew up") + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 3, 200L, noopRetryHook, delays += _) { + attempts += 1 + throw cause + } + } + assert(attempts == 3) + assert(delays.toList == List(200L, 400L)) + assert(failure.getCause eq cause) + } +} diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala index 8e92ecb941..e79bf63b1d 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala @@ -24,6 +24,7 @@ import io.lakefs.clients.sdk._ import io.lakefs.clients.sdk.model.ResetCreation.TypeEnum import io.lakefs.clients.sdk.model._ import org.apache.texera.common.config.StorageConfig +import org.apache.texera.common.util.RetryUtil import java.io.{File, FileOutputStream, InputStream} import java.net.URI @@ -75,53 +76,16 @@ object LakeFSStorageClient extends LazyLogging { private val branchName: String = "main" def healthCheck(): Unit = { - retryWithBackoff(HealthCheckMaxAttempts, HealthCheckInitialDelayMillis) { + RetryUtil.withBackoff( + description = "connect to lake fs server", + maxAttempts = HealthCheckMaxAttempts, + initialDelayMillis = HealthCheckInitialDelayMillis, + onRetry = attempt => logger.warn(attempt.message) + ) { this.healthCheckApi.healthCheck().execute() } } - /** - * Runs `operation`, retrying on failure with exponential backoff (the delay - * doubles after each failed attempt) until it succeeds or `maxAttempts` is - * reached. The final failure is rethrown with the last exception as its cause. - * If interrupted while waiting, restores the interrupt status and fails fast. - * - * `sleep` is injectable so the backoff can be exercised in tests without real waiting. - */ - private[util] def retryWithBackoff( - maxAttempts: Int, - initialDelayMillis: Long, - sleep: Long => Unit = Thread.sleep - )(operation: => Unit): Unit = { - var attempt = 1 - var delayMillis = initialDelayMillis - while (true) { - try { - operation - return - } catch { - case ie: InterruptedException => - // Restore the interrupt status and fail fast rather than retrying. - Thread.currentThread().interrupt() - throw new RuntimeException("Interrupted while waiting to retry lake fs health check", ie) - case e: Exception => - if (attempt >= maxAttempts) { - throw new RuntimeException( - s"Failed to connect to lake fs server after $maxAttempts attempts: ${e.getMessage}", - e - ) - } - logger.warn( - s"LakeFS not reachable (attempt $attempt/$maxAttempts): ${e.getMessage}. " + - s"Retrying in ${delayMillis}ms..." - ) - sleep(delayMillis) - attempt += 1 - delayMillis *= 2 - } - } - } - /** * Initializes a new repository in LakeFS. * diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala index ff5dca57d9..ad2bb38977 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala @@ -21,55 +21,10 @@ package org.apache.texera.amber.core.storage.util import org.scalatest.flatspec.AnyFlatSpec -import scala.collection.mutable.ListBuffer - class LakeFSStorageClientSpec extends AnyFlatSpec { - "retryWithBackoff" should "run the operation once and not sleep when it succeeds immediately" in { - var attempts = 0 - val delays = ListBuffer.empty[Long] - LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) { - attempts += 1 - } - assert(attempts == 1) - assert(delays.isEmpty) - } - - it should "retry until success and double the delay after each failed attempt" in { - var attempts = 0 - val delays = ListBuffer.empty[Long] - LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) { - attempts += 1 - if (attempts < 3) throw new RuntimeException("transient") - } - assert(attempts == 3) - assert(delays.toList == List(200L, 400L)) - } - - it should "give up after maxAttempts and preserve the last failure as the cause" in { - var attempts = 0 - val cause = new RuntimeException("still down") - val ex = intercept[RuntimeException] { - LakeFSStorageClient.retryWithBackoff(3, 200L, _ => ()) { - attempts += 1 - throw cause - } - } - assert(attempts == 3) - assert(ex.getMessage.contains("after 3 attempts")) - assert(ex.getCause eq cause) - } - - it should "fail fast and restore the interrupt status when interrupted" in { - val ex = intercept[RuntimeException] { - LakeFSStorageClient.retryWithBackoff(5, 200L, _ => ()) { - throw new InterruptedException("interrupted") - } - } - // Thread.interrupted() both reads and clears the flag, so the interrupt was restored. - assert(Thread.interrupted()) - assert(ex.getCause.isInstanceOf[InterruptedException]) - } + // `healthCheck` retries through the shared `RetryUtil.withBackoff`; that contract (progression, + // give-up wrapping, interrupt fail-fast) is covered by `RetryUtilSpec` in `common/util`. "parsePhysicalAddress" should "split a well-formed address into bucket and key" in { assert( diff --git a/file-service/src/main/scala/org/apache/texera/service/FileService.scala b/file-service/src/main/scala/org/apache/texera/service/FileService.scala index d8e3ff122d..1bb29f5dab 100644 --- a/file-service/src/main/scala/org/apache/texera/service/FileService.scala +++ b/file-service/src/main/scala/org/apache/texera/service/FileService.scala @@ -26,6 +26,7 @@ import io.dropwizard.configuration.{EnvironmentVariableSubstitutor, Substituting import io.dropwizard.core.Application import io.dropwizard.core.setup.{Bootstrap, Environment} import org.apache.texera.common.config.StorageConfig +import org.apache.texera.common.util.RetryUtil import org.apache.texera.amber.core.storage.util.LakeFSStorageClient import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer} import org.apache.texera.dao.SqlServer @@ -70,11 +71,11 @@ class FileService extends Application[FileServiceConfiguration] with LazyLogging ) // check if the texera dataset bucket exists, if not create it - awaitDependency("texera dataset bucket") { + awaitDependency("reach the texera dataset bucket") { S3StorageClient.createBucketIfNotExist(StorageConfig.lakefsBucketName) } // ensure the large-binary S3 bucket exists before any workflow execution attempts to use it - awaitDependency("large-binary bucket") { + awaitDependency("reach the large-binary bucket") { S3StorageClient.createBucketIfNotExist(LargeBinaryManager.DEFAULT_BUCKET) } // check if we can connect to the lakeFS service @@ -121,52 +122,24 @@ class FileService extends Application[FileServiceConfiguration] with LazyLogging .manage(new StagedFileCleanupJob(retentionHours, intervalMinutes)) /** - * Runs `operation`, retrying with exponential backoff until it succeeds or `maxAttempts` is - * reached, to tolerate a slow-to-start object store. The last failure is rethrown as the cause. - * `sleep` is injectable for tests. Defaults: 6 attempts from 200ms (200, 400, 800, 1600, 3200), ~6s. + * Waits for a startup dependency (a slow-to-start object store) via the shared backoff retry, + * logging each retry under this service's logger. `description` is a verb phrase, e.g. + * "reach the texera dataset bucket". `sleep` is injectable for tests. + * Defaults: 6 attempts from 200ms (200, 400, 800, 1600, 3200), ~6s. */ private[service] def awaitDependency( description: String, maxAttempts: Int = 6, initialDelayMillis: Long = 200L, sleep: Long => Unit = Thread.sleep - )(operation: => Unit): Unit = { - // Restore the interrupt status and fail fast rather than retrying, whether the - // interrupt arrives while running `operation` or while sleeping between attempts. - def failInterrupted(ie: InterruptedException): Nothing = { - Thread.currentThread().interrupt() - throw new RuntimeException(s"Interrupted while waiting for $description", ie) - } - - var attempt = 1 - var delayMillis = initialDelayMillis - while (true) { - try { - operation - return - } catch { - case ie: InterruptedException => failInterrupted(ie) - case e: Exception => - if (attempt >= maxAttempts) { - throw new RuntimeException( - s"$description not ready after $maxAttempts attempts: ${e.getMessage}", - e - ) - } - logger.warn( - s"$description not ready (attempt $attempt/$maxAttempts): ${e.getMessage}. " + - s"Retrying in ${delayMillis}ms..." - ) - try { - sleep(delayMillis) - } catch { - case ie: InterruptedException => failInterrupted(ie) - } - attempt += 1 - delayMillis *= 2 - } - } - } + )(operation: => Unit): Unit = + RetryUtil.withBackoff( + description, + maxAttempts, + initialDelayMillis, + attempt => logger.warn(attempt.message), + sleep + )(operation) } object FileService { diff --git a/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala index 3deb60b6f7..71606c5236 100644 --- a/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala @@ -34,7 +34,7 @@ class FileServiceSpec extends AnyFlatSpec { "awaitDependency" should "run the operation once and not sleep when it succeeds immediately" in { var attempts = 0 val delays = ListBuffer.empty[Long] - service.awaitDependency("dep", 6, 200L, delays += _) { + service.awaitDependency("reach dep", 6, 200L, delays += _) { attempts += 1 } assert(attempts == 1) @@ -45,7 +45,7 @@ class FileServiceSpec extends AnyFlatSpec { // Exercises the default maxAttempts/initialDelay/sleep parameters: a first-try success // returns without ever invoking the (real Thread.sleep) default backoff. var attempts = 0 - service.awaitDependency("dep") { + service.awaitDependency("reach dep") { attempts += 1 } assert(attempts == 1) @@ -54,7 +54,7 @@ class FileServiceSpec extends AnyFlatSpec { it should "retry until success and double the delay after each failed attempt" in { var attempts = 0 val delays = ListBuffer.empty[Long] - service.awaitDependency("dep", 6, 200L, delays += _) { + service.awaitDependency("reach dep", 6, 200L, delays += _) { attempts += 1 if (attempts < 3) throw new RuntimeException("not reachable yet") } @@ -66,7 +66,7 @@ class FileServiceSpec extends AnyFlatSpec { var attempts = 0 val delays = ListBuffer.empty[Long] val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 6, 200L, delays += _) { + service.awaitDependency("reach dep", 6, 200L, delays += _) { attempts += 1 throw new RuntimeException("down") } @@ -81,7 +81,7 @@ class FileServiceSpec extends AnyFlatSpec { var attempts = 0 val cause = new RuntimeException("still down") val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 3, 200L, _ => ()) { + service.awaitDependency("reach dep", 3, 200L, _ => ()) { attempts += 1 throw cause } @@ -92,96 +92,35 @@ class FileServiceSpec extends AnyFlatSpec { assert(ex.getCause eq cause) } - it should "give up immediately without sleeping when maxAttempts is 1" in { - var attempts = 0 - val delays = ListBuffer.empty[Long] - val cause = new RuntimeException("still down") - val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 1, 200L, delays += _) { - attempts += 1 - throw cause - } - } - assert(attempts == 1) - assert(delays.isEmpty) - assert(ex.getMessage.contains("after 1 attempts")) - assert(ex.getCause eq cause) - } - it should "fail fast and restore the interrupt status when the operation is interrupted" in { val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 6, 200L, _ => ()) { + service.awaitDependency("reach dep", 6, 200L, _ => ()) { throw new InterruptedException("interrupted") } } // Thread.interrupted() both reads and clears the flag, so the interrupt was restored. assert(Thread.interrupted()) - assert(ex.getMessage.contains("Interrupted while waiting for dep")) - assert(ex.getCause.isInstanceOf[InterruptedException]) - } - - it should "fail fast and restore the interrupt status when interrupted while sleeping between attempts" in { - var attempts = 0 - val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 6, 200L, _ => throw new InterruptedException("interrupted")) { - attempts += 1 - throw new RuntimeException("not reachable yet") - } - } - // The operation failed once, then the interrupt arrived during the backoff sleep. - assert(attempts == 1) - // Thread.interrupted() both reads and clears the flag, so the interrupt was restored. - assert(Thread.interrupted()) - assert(ex.getMessage.contains("Interrupted while waiting for dep")) + assert(ex.getMessage.contains("Interrupted while waiting to reach dep")) assert(ex.getCause.isInstanceOf[InterruptedException]) } - it should "succeed on the final allowed attempt without giving up one try too early" in { - // Boundary for `attempt >= maxAttempts`: the operation only succeeds on the very last - // attempt, so the loop must not give up prematurely. Expect maxAttempts - 1 backoff waits. - var attempts = 0 - val delays = ListBuffer.empty[Long] - service.awaitDependency("dep", 3, 200L, delays += _) { - attempts += 1 - if (attempts < 3) throw new RuntimeException("not reachable yet") - } - assert(attempts == 3) - assert(delays.toList == List(200L, 400L)) - } - - it should "honor a custom initial delay when computing the backoff progression" in { - // Guards against the initial delay being hardcoded: starting from 50ms the geometric - // progression must be 50, 100, 200 rather than the default 200-based sequence. - var attempts = 0 - val delays = ListBuffer.empty[Long] - val ex = intercept[RuntimeException] { - service.awaitDependency("dep", 4, 50L, delays += _) { - attempts += 1 - throw new RuntimeException("down") - } - } - assert(attempts == 4) - assert(delays.toList == List(50L, 100L, 200L)) - assert(ex.getMessage.contains("after 4 attempts")) - } - - it should "include the underlying failure message when giving up" in { + it should "include the description and the underlying failure message when giving up" in { val ex = intercept[RuntimeException] { - service.awaitDependency("dataset bucket", 2, 200L, _ => ()) { + service.awaitDependency("reach the dataset bucket", 2, 200L, _ => ()) { throw new RuntimeException("connection refused") } } - assert(ex.getMessage.contains("dataset bucket not ready after 2 attempts")) + assert(ex.getMessage.contains("Failed to reach the dataset bucket after 2 attempts")) assert(ex.getMessage.contains("connection refused")) } - it should "propagate a non-Exception Throwable immediately without retrying or wrapping it" in { - // The catch clause only matches Exception, so an Error must escape on the first attempt: - // it is neither retried nor wrapped in the \"not ready after N attempts\" RuntimeException. + it should "propagate a fatal Throwable immediately without retrying or wrapping it" in { + // A fatal throwable is not transient, so it must escape on the first attempt: it is neither + // retried nor wrapped in the "Failed to ... after N attempts" RuntimeException. var attempts = 0 val delays = ListBuffer.empty[Long] val err = intercept[StackOverflowError] { - service.awaitDependency("dep", 6, 200L, delays += _) { + service.awaitDependency("reach dep", 6, 200L, delays += _) { attempts += 1 throw new StackOverflowError("boom") }
