Yicong-Huang commented on code in PR #7119:
URL: https://github.com/apache/texera/pull/7119#discussion_r3680531559
##########
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)
Review Comment:
The new module needs one more registration:
`.github/workflows/build.yml:308-316` lists the common modules for the
test-running `jacoco` task by hand, and `Util` isn't in it. The comment right
above that list says why the list can't be implicit — sbt's `test` does not
transit `dependsOn` — so `WorkflowCore/jacoco` will not reach `Util`'s Test
config. Every other sbt call in CI is scoped to a named project (`:648`,
`:660`, `:796`, `:932`) and there is no root `sbt test`, so `RetryUtilSpec`'s
10 cases never execute.
That matters more than "new tests unguarded", because this PR also deletes 4
retry tests from `LakeFSStorageClientSpec` (ran under `WorkflowCore/jacoco`)
and 4 from `FileServiceSpec` (ran under the `FileService/jacoco` matrix leg).
Net -8 CI-executed tests on this behavior, and the
interrupt-during-backoff-sleep fix you highlight now has no CI guard at all —
where `FileServiceSpec` had one before. Codecov agrees from the other side: no
`util` flag, and `Files 1159 -> 1159` despite three added files.
Adding `"Util/jacoco" \` to that list fixes it; the codecov upload glob
already picks the report up. Commenting here because the workflow file isn't in
the diff.
##########
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)
Review Comment:
`NonFatal` here is wider than the `case e: Exception` both replaced loops
used, so both call sites now retry non-fatal `Error`s — `java.io.IOError`,
`ServiceConfigurationError`, `AssertionError`. On an S3/LakeFS startup path
those are reachable, and they go from propagating immediately to costing ~6s of
backoff and arriving wrapped.
I think `NonFatal` is the right choice — it's parity with `Utils.retry`,
which is the stated point of the pair — but nothing currently records it as a
decision or guards it. Both fatal-throwable cases sit on the other side of the
line: `RetryUtilSpec:177` uses `ControlThrowable` and `FileServiceSpec:125`
uses `StackOverflowError`, and both were fatal under `Exception` too, so
neither can observe the change.
One `RetryUtilSpec` case asserting that a non-fatal `Error` **is** retried
would pin the new contract, plus a line in the description noting the widening.
##########
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
Review Comment:
The rule is right but the premise stated for it isn't: `Auth`, `Config`,
`Resource`, `DAO` and `PyBuilder` have no `dependsOn(Util)` edge, and
`ConfigService`, `AccessControlService` and `NotebookMigrationService` never
reach it even transitively. Same overstatement at `build.sbt:120` and in
`RetryUtil.scala:28` ("Every module can reach this one").
```suggestion
// 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
```
##########
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.
Review Comment:
`LakeFSStorageClient.retryWithBackoff` doesn't exist after this PR, so a
reader who greps for it finds nothing. The paragraph also only makes sense to
someone holding the PR's history ("this util replaced", "the LakeFS copy") —
same at `:152` and in `LakeFSStorageClientSpec:26` ("the loop that used to live
here"). Stating the contract directly survives the merge:
```suggestion
* Coverage is the full contract both blocking callers rely on: the
doubling progression, the
* give-up wrapping, and interrupt fail-fast during the operation and
during a backoff sleep.
```
##########
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.
Review Comment:
This PR makes the comment 24 lines below (`:97`) false: it justifies
`NonFatal` here by saying it "matches how the blocking backoff loops elsewhere
in the repo catch `Exception`". After the consolidation there is one blocking
loop, and it catches `NonFatal`.
Worth correcting rather than deleting, because the fixed version is a
stronger claim than the original — the two halves now agree exactly:
```
// `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.
```
##########
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 = _ => ()
Review Comment:
`noRetryHook` reads as "a hook that disables retrying", which is confusing
in the cases that are actually about not retrying (`:102` maxAttempts = 1,
`:173` fatal throwable) — this hook fires on every retry, it just does nothing.
`noopRetryHook` or `ignoreRetries` would say that. Eight call sites, so not a
one-line suggestion.
##########
file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala:
##########
@@ -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.
Review Comment:
The `\"` sequences are literal in the file — nothing needs escaping in a
Scala line comment.
```suggestion
// neither retried nor wrapped in the "Failed to ... after N attempts"
RuntimeException.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
