aglinxinyuan commented on code in PR #7119:
URL: https://github.com/apache/texera/pull/7119#discussion_r3680898942
##########
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:
Confirmed and fixed in 93fb507 — `"Util/jacoco"` added to that list.
Verified locally that the task both runs the spec and writes
`common/util/target/scala-2.13/jacoco/report/jacoco.xml`, so the upload glob
picks it up exactly as you said, no codecov change needed.
The net -8 framing is the part I'd have missed: the
interrupt-during-backoff-sleep behavior is the one thing this PR actually
fixes, and it would have shipped with no CI guard at all after
`FileServiceSpec` lost its copy. Thanks for tracing every sbt invocation rather
than just flagging the missing entry.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Fixed in 93fb507 with your wording — you're right that it's a stronger claim
now that there is exactly one blocking loop to point at.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Agreed on all three points, and this was the one real behavior change hiding
in a "structure only" refactor.
Added a `RetryUtilSpec` case asserting a non-fatal `Error` **is** retried
(`AssertionError`, 3 attempts, 200/400 progression, wrapped with the cause
preserved), so the boundary is now pinned from both sides rather than only the
fatal side. Also recorded the widening in the scaladoc and in a dedicated
paragraph in the description, including the consequence you named: on the
S3/LakeFS startup paths such a failure now costs the full backoff before
surfacing.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Right, overstated — applied your suggestion, and fixed the same claim in the
two other places it had spread to (`build.sbt:120` and `RetryUtil.scala`'s
scaladoc).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Applied. Also rewrote the interrupt-during-sleep comment at `:152` and the
note in `LakeFSStorageClientSpec` the same way, so none of the three depend on
knowing what this PR replaced — the `:152` one now states the reason the case
exists (a `catch` cannot catch what its own body throws) instead of whose copy
got it wrong.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Fixed — those escapes rode along when I moved the comment. Unescaped.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Renamed to `noopRetryHook` across all 8 sites — agreed that "no retry hook"
reads exactly backwards in the two cases that are about not retrying.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
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]
