From 45d0401babf1ed41bb34e53ee94619ed80b5770a Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Wed, 29 Jul 2026 22:44:31 -0700 Subject: [PATCH 1/2] refactor(common): host one exponential-backoff retry in common/util The same doubling-backoff loop was written three times: the async one in amber's Utils, plus two structurally identical blocking copies in LakeFSStorageClient and FileService that no module could share because amber's Utils sits above them in the module graph. Add a dependency-free `common/util` module with `RetryUtil.withBackoff` and point both blocking copies at it. The util owns the attempt counting, the doubling, what counts as transient (NonFatal), interrupt fail-fast with the interrupt status restored, and the message wording; callers pass a verb-phrase description and log retries through their own logger via the `onRetry` hook, so warnings stay attributed to the caller. This also fixes a hole in the LakeFS copy: its `sleep` call sat inside the `catch` block, so an interrupt raised while waiting escaped raw with the interrupt flag left cleared. FileService's copy already handled that; the shared util keeps the better behavior for both. The module is deliberately dependency-free -- everything depends on it, so anything added there lands on every service's classpath. That is also why the non-blocking variant stays in amber: it needs com.twitter:util-core, which only amber declares. The two now cross-reference each other. Closes #7095 --- AGENTS.md | 2 +- .../texera/amber/engine/common/Utils.scala | 5 + build.sbt | 8 +- common/util/build.sbt | 56 ++++++ .../apache/texera/common/util/RetryUtil.scala | 114 +++++++++++ .../texera/common/util/RetryUtilSpec.scala | 187 ++++++++++++++++++ .../storage/util/LakeFSStorageClient.scala | 50 +---- .../util/LakeFSStorageClientSpec.scala | 51 +---- .../apache/texera/service/FileService.scala | 57 ++---- .../texera/service/FileServiceSpec.scala | 89 ++------- 10 files changed, 409 insertions(+), 210 deletions(-) create mode 100644 common/util/build.sbt create mode 100644 common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala create mode 100644 common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala diff --git a/AGENTS.md b/AGENTS.md index b21ed6d6a1a..7118073a781 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 5dee2c67546..a11b208f518 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. * diff --git a/build.sbt b/build.sbt index 720fc28b806..513df3e78c6 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 lands on every service's classpath. +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 00000000000..e34e56476d8 --- /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: +// every other module depends on it, so anything added here lands on every +// service's classpath. 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 00000000000..b850fb22ca5 --- /dev/null +++ b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala @@ -0,0 +1,114 @@ +/* + * 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. + * + * Every module can reach this one, so a new retry loop should not be written 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. 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 00000000000..fea23b54635 --- /dev/null +++ b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala @@ -0,0 +1,187 @@ +/* + * 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. + * + * These cases are the union of what the two hand-rolled loops this util replaced were tested for + * (`LakeFSStorageClient.retryWithBackoff`, `FileService.awaitDependency`), plus the interrupt + * during a backoff sleep, which the LakeFS copy did not handle. + */ +class RetryUtilSpec extends AnyFlatSpec { + + private def noRetryHook: 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, noRetryHook, 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, noRetryHook, 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, noRetryHook, 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, noRetryHook, 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, noRetryHook, _ => ()) { + 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, noRetryHook, 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, noRetryHook, 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 { + // The hole in the LakeFS copy: its `sleep` sat inside the `catch`, so an interrupt raised + // while waiting escaped raw, with the interrupt flag left cleared. + var attempts = 0 + val failure = intercept[RuntimeException] { + RetryUtil.withBackoff( + "reach the store", + 5, + 200L, + noRetryHook, + _ => 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, noRetryHook, delays += _) { + attempts += 1 + throw Fatal + } + } + assert(attempts == 1) + assert(delays.isEmpty) + } +} 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 8e92ecb9415..e79bf63b1df 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 ff5dca57d99..1758462b7c4 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,12 @@ 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]) - } + // The health-check backoff now comes from the shared `RetryUtil.withBackoff`, whose contract + // (progression, give-up wrapping, interrupt handling) is covered by `RetryUtilSpec` in + // `common/util` -- including an interrupt during the backoff sleep, which the loop that used to + // live here let escape without restoring the interrupt status. "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 d8e3ff122d5..1bb29f5dab3 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 3deb60b6f76..746f4089098 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 { + // Only NonFatal failures are transient, so an Error 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") } From 93fb5071e6a4eac804c783bd70c0bb44317ed027 Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Thu, 30 Jul 2026 00:39:06 -0700 Subject: [PATCH 2/2] fix(ci): run the new common/util module's tests, and address review notes The common-module list that `build.yml` hands to `jacoco` is maintained by hand -- sbt's `test` does not transit `dependsOn`, so nothing else reaches a module's Test config -- and `Util` was missing from it. `RetryUtilSpec` never ran in CI, while this PR removed 8 tests that did, including the only guard on the interrupt-during-backoff-sleep behavior. Register `Util/jacoco`; the codecov upload glob already picks the report up. Also from review: - pin the one behavior change the consolidation carries -- `NonFatal` is wider than the `case e: Exception` both loops used, so a non-fatal `Error` is now retried rather than propagated -- with a test and a scaladoc note. - correct the claim that every module depends on `common/util`: five common modules and three services have no edge to it. - state the spec's contract directly instead of describing what it replaced, so the docs survive this PR's history. - fix the amber-side comment this PR made false, unescape quotes in a Scala line comment, and rename `noRetryHook` -> `noopRetryHook`. --- .github/workflows/build.yml | 1 + .../texera/amber/engine/common/Utils.scala | 4 +- build.sbt | 2 +- common/util/build.sbt | 4 +- .../apache/texera/common/util/RetryUtil.scala | 14 ++++-- .../texera/common/util/RetryUtilSpec.scala | 49 +++++++++++++------ .../util/LakeFSStorageClientSpec.scala | 6 +-- .../texera/service/FileServiceSpec.scala | 4 +- 8 files changed, 53 insertions(+), 31 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index becd8deb34c..a82cbfde95b 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/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 a11b208f518..393e5b343a5 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 @@ -93,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 513df3e78c6..b10661fd70e 100644 --- a/build.sbt +++ b/build.sbt @@ -118,7 +118,7 @@ val nettyDependencyOverrides = Seq( ThisBuild / excludeDependencies += ExclusionRule("log4j", "log4j") // Dependency-free helpers (retry/backoff, ...) that any module may depend on. Keep it that way: -// anything added here lands on every service's classpath. +// 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) diff --git a/common/util/build.sbt b/common/util/build.sbt index e34e56476d8..0f4446edccd 100644 --- a/common/util/build.sbt +++ b/common/util/build.sbt @@ -48,8 +48,8 @@ Compile / scalacOptions ++= Seq( ///////////////////////////////////////////////////////////////////////////// // This module is deliberately dependency-free apart from the test framework: -// every other module depends on it, so anything added here lands on every -// service's classpath. Callers pass their own logger in through the hooks +// 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 index b850fb22ca5..3ebf094b4e7 100644 --- 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 @@ -25,8 +25,9 @@ import scala.util.control.NonFatal /** * Retry with exponential backoff, for blocking work. * - * Every module can reach this one, so a new retry loop should not be written by hand. If the work - * returns a `Future` rather than blocking, use the non-blocking sibling + * 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. @@ -54,9 +55,12 @@ object RetryUtil { * 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. 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. + * 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 ...". 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 index fea23b54635..828cb8d01e5 100644 --- 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 @@ -29,18 +29,18 @@ 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. * - * These cases are the union of what the two hand-rolled loops this util replaced were tested for - * (`LakeFSStorageClient.retryWithBackoff`, `FileService.awaitDependency`), plus the interrupt - * during a backoff sleep, which the LakeFS copy did not handle. + * 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 noRetryHook: RetryAttempt => Unit = _ => () + 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, noRetryHook, delays += _) { + val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { attempts += 1 "value" } @@ -52,7 +52,7 @@ class RetryUtilSpec extends AnyFlatSpec { 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, noRetryHook, delays += _) { + val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { attempts += 1 if (attempts < 3) throw new RuntimeException("transient") attempts @@ -65,7 +65,7 @@ class RetryUtilSpec extends AnyFlatSpec { // 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, noRetryHook, delays += _) { + RetryUtil.withBackoff("reach the store", 4, 50L, noopRetryHook, delays += _) { throw new RuntimeException("down") } } @@ -76,7 +76,7 @@ class RetryUtilSpec extends AnyFlatSpec { // 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, noRetryHook, delays += _) { + RetryUtil.withBackoff("reach the store", 3, 200L, noopRetryHook, delays += _) { attempts += 1 if (attempts < 3) throw new RuntimeException("transient") } @@ -88,7 +88,7 @@ class RetryUtilSpec extends AnyFlatSpec { val cause = new RuntimeException("still down") var attempts = 0 val failure = intercept[RuntimeException] { - RetryUtil.withBackoff("connect to lake fs server", 3, 200L, noRetryHook, _ => ()) { + RetryUtil.withBackoff("connect to lake fs server", 3, 200L, noopRetryHook, _ => ()) { attempts += 1 throw cause } @@ -102,7 +102,7 @@ class RetryUtilSpec extends AnyFlatSpec { val delays = ListBuffer.empty[Long] var attempts = 0 val failure = intercept[RuntimeException] { - RetryUtil.withBackoff("reach the store", 1, 200L, noRetryHook, delays += _) { + RetryUtil.withBackoff("reach the store", 1, 200L, noopRetryHook, delays += _) { attempts += 1 throw new RuntimeException("still down") } @@ -136,7 +136,7 @@ class RetryUtilSpec extends AnyFlatSpec { 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, noRetryHook, delays += _) { + RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { throw new InterruptedException("interrupted") } } @@ -148,15 +148,15 @@ class RetryUtilSpec extends AnyFlatSpec { } it should "fail fast and restore the interrupt status when interrupted during a backoff sleep" in { - // The hole in the LakeFS copy: its `sleep` sat inside the `catch`, so an interrupt raised - // while waiting escaped raw, with the interrupt flag left cleared. + // 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, - noRetryHook, + noopRetryHook, _ => throw new InterruptedException("interrupted") ) { attempts += 1 @@ -176,7 +176,7 @@ class RetryUtilSpec extends AnyFlatSpec { var attempts = 0 object Fatal extends ControlThrowable intercept[ControlThrowable] { - RetryUtil.withBackoff("reach the store", 5, 200L, noRetryHook, delays += _) { + RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) { attempts += 1 throw Fatal } @@ -184,4 +184,23 @@ class RetryUtilSpec extends AnyFlatSpec { 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/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 1758462b7c4..ad2bb389775 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 @@ -23,10 +23,8 @@ import org.scalatest.flatspec.AnyFlatSpec class LakeFSStorageClientSpec extends AnyFlatSpec { - // The health-check backoff now comes from the shared `RetryUtil.withBackoff`, whose contract - // (progression, give-up wrapping, interrupt handling) is covered by `RetryUtilSpec` in - // `common/util` -- including an interrupt during the backoff sleep, which the loop that used to - // live here let escape without restoring the interrupt status. + // `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/test/scala/org/apache/texera/service/FileServiceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala index 746f4089098..71606c52360 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 @@ -115,8 +115,8 @@ class FileServiceSpec extends AnyFlatSpec { } it should "propagate a fatal Throwable immediately without retrying or wrapping it" in { - // Only NonFatal failures are transient, so an Error must escape on the first attempt: it is - // neither retried nor wrapped in the \"Failed to ... after N attempts\" RuntimeException. + // 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] {