refactor(util): host one exponential-backoff retry #7119
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
aglinxinyuan marked this conversation as resolved.
|
||
| } | ||
|
|
||
| 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) | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.