chore(amber): bound region termination to a short backoff - #7088
Merged
aglinxinyuan merged 6 commits intoJul 30, 2026
Conversation
Region termination retried EndWorker + gracefulStop up to 150 times at a flat 200ms, holding a region's teardown for ~30s before a failed kill surfaced. Killing a worker is deterministic, so retries only need to ride out a transient failure: one attempt plus three backed-off retries at 200/400/800 ms (~1.4s). The backoff schedule is now the retry budget (one retry per delay), so the attempt count and the delays cannot drift apart. Give-up behavior is unchanged: it still fails loudly, naming every worker that never terminated. Closes apache#7087
Contributor
Automated Reviewer SuggestionsBased on the
|
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 373 | 0.228 | 25,758/40,261/40,261 us | 🔴 +20.1% / 🔴 +155.2% |
| 🔴 | bs=100 sw=10 sl=64 | 735 | 0.449 | 130,082/214,383/214,383 us | 🔴 +46.3% / 🔴 +99.7% |
| ⚪ | bs=1000 sw=10 sl=64 | 921 | 0.562 | 1,089,133/1,125,310/1,125,310 us | ⚪ within ±5% / 🔴 +11.1% |
Baseline details
Latest main f28d8ec from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 373 tuples/sec | 413 tuples/sec | 786.12 tuples/sec | -9.7% | -52.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.228 MB/s | 0.252 MB/s | 0.48 MB/s | -9.5% | -52.5% |
| bs=10 sw=10 sl=64 | p50 | 25,758 us | 23,651 us | 12,305 us | +8.9% | +109.3% |
| bs=10 sw=10 sl=64 | p95 | 40,261 us | 33,536 us | 15,774 us | +20.1% | +155.2% |
| bs=10 sw=10 sl=64 | p99 | 40,261 us | 33,536 us | 18,978 us | +20.1% | +112.1% |
| bs=100 sw=10 sl=64 | throughput | 735 tuples/sec | 815 tuples/sec | 999.71 tuples/sec | -9.8% | -26.5% |
| bs=100 sw=10 sl=64 | MB/s | 0.449 MB/s | 0.497 MB/s | 0.61 MB/s | -9.7% | -26.4% |
| bs=100 sw=10 sl=64 | p50 | 130,082 us | 122,252 us | 100,616 us | +6.4% | +29.3% |
| bs=100 sw=10 sl=64 | p95 | 214,383 us | 146,554 us | 107,356 us | +46.3% | +99.7% |
| bs=100 sw=10 sl=64 | p99 | 214,383 us | 146,554 us | 113,255 us | +46.3% | +89.3% |
| bs=1000 sw=10 sl=64 | throughput | 921 tuples/sec | 915 tuples/sec | 1,031 tuples/sec | +0.7% | -10.7% |
| bs=1000 sw=10 sl=64 | MB/s | 0.562 MB/s | 0.559 MB/s | 0.63 MB/s | +0.5% | -10.7% |
| bs=1000 sw=10 sl=64 | p50 | 1,089,133 us | 1,089,277 us | 980,328 us | -0.0% | +11.1% |
| bs=1000 sw=10 sl=64 | p95 | 1,125,310 us | 1,151,335 us | 1,027,528 us | -2.3% | +9.5% |
| bs=1000 sw=10 sl=64 | p99 | 1,125,310 us | 1,151,335 us | 1,054,298 us | -2.3% | +6.7% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,535.93,200,128000,373,0.228,25758.29,40260.57,40260.57
1,100,10,64,20,2721.08,2000,1280000,735,0.449,130082.34,214383.23,214383.23
2,1000,10,64,20,21714.44,20000,12800000,921,0.562,1089133.07,1125310.07,1125310.07
aglinxinyuan
changed the title
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7088 +/- ##
============================================
- Coverage 79.38% 79.38% -0.01%
+ Complexity 3799 3795 -4
============================================
Files 1159 1159
Lines 46150 46153 +3
Branches 5129 5128 -1
============================================
+ Hits 36638 36640 +2
- Misses 7890 7892 +2
+ Partials 1622 1621 -1
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Region termination carried its own backoff loop. `Utils.retry` already covers blocking work, but its `Thread.sleep` would stall the coordinator thread, so the async path could not reuse it. Add `Utils.retryAsync`: the same attempts-and-doubling-backoff contract for `Future`-returning logic, waiting on a `Timer` instead of blocking a thread, with an `onRetry` hook so callers keep their own log context. Region termination now configures it (4 attempts from a 200ms base) instead of hand-rolling the loop, leaving one retry util for new code to reuse. Addresses review feedback on apache#7088.
The blocking `Utils.retry` had no production callers -- only its own tests -- so keeping it beside the new async variant would have shipped two near-identical retries. Replace it: `Utils.retry` now takes the same `attempts` + `baseBackoffTimeInMS`-doubling knobs but schedules its waits on a `Timer`, so it can be used from an actor or coordinator thread where `Thread.sleep` would stall unrelated work. Region termination is its first real caller. A blocking front door can come back as a thin sibling if a caller needs one.
Closed
1 task
Contributor
Author
|
Good call -- done, and
Utils.retry(attempts, baseBackoffTimeInMS, timer, onRetry) { operationReturningAFuture }One thing to flag, since you asked to keep On "lots of places have this logic": the other two are |
Yicong-Huang
left a comment
Yicong-Huang
left a comment
Contributor
There was a problem hiding this comment.
🟡 0 must-fix · 2 advisory · 0 polish — clean, well-scoped refactor with exact, non-flaky tests. Two judgment-call notes for you.
Correctness (1)
Utils.scala:90— Scaladoc "Fatal errors are not retried" vs. arescuethat matches anyThrowable(advisory, see inline)
Conventions (1)
- Title type
choreunderstates a runtime failure-path change that closes a Performance issue (#7087) — considerperf(amber)/fix(amber). Note:backport-auto-label.ymllabels onlyfix(...)PRs for backport, sochoresilently excludes this from any release branch. Your call on type and backport intent. (advisory)
Verification trace
Treated the retry rewrite as an equivalence claim. First attempt: Utils.retry evaluates terminateWorkers synchronously via Future(fn).flatten on the calling thread — same as the old direct .rescue call, only enqueues RPC futures, no new blocking. Retries run on the Timer thread via Future.sleep(...)(timer).flatMap(...) — the same thread the old code used, so the #6923 window shrinks but is not newly opened. Budget: attempts 1–3 retry, attempt 4 propagates; waits 200/400/800; onRetry fires 3× and never after the last attempt. Give-up equivalence: old message used the reached attempt, new uses maxTerminationAttempts; since retry always exhausts to attempts, reached == budget at give-up, and <= 1 ? "1 attempt" also covers attempts=0. The teardown sequence in terminateWorkers is byte-unchanged, so the teardown-order invariant holds. Only production caller of Utils.retry is this manager, so the signature break leaves no stale caller.
Yicong-Huang
approved these changes
Jul 30, 2026
Yicong-Huang
left a comment
Yicong-Huang
left a comment
Contributor
There was a problem hiding this comment.
Approving — clean, well-scoped refactor; the two earlier notes are non-blocking judgment calls. Nice test rig (exact, non-flaky backoff assertions via the inline timer).
The rescue guard keyed only on the attempt count, so a fatal error handed back as a failed `Future` was retried -- contradicting the scaladoc, and inconsistent with the synchronous path, where `Future(fn)`'s `Try` already filters fatals out. Guard with `NonFatal` so both shapes propagate on the first attempt, matching how the blocking backoff loops elsewhere in the repo catch `Exception`. Addresses review feedback on apache#7088.
Contributor
Author
|
On the title: Both notes are addressed now ( |
renovate-bot
pushed a commit
to renovate-bot/apache-_-texera
that referenced
this pull request
Jul 30, 2026
### What changes were proposed in this PR?
The same doubling-backoff loop was written three times, and no module
could share the others' copy: amber's `Utils` sits above
`common/workflow-core` and `file-service` in the module graph.
New dependency-free **`common/util`** module with
`RetryUtil.withBackoff`, and both blocking copies now call it:
| Site | Before | After |
| --- | --- | --- |
| `LakeFSStorageClient.healthCheck` | private 27-line `retryWithBackoff`
| `RetryUtil.withBackoff("connect to lake fs server", 5, 200, ...)` |
| `FileService.awaitDependency` | private 36-line loop | 7-line
delegation keeping its 6/200 defaults + logger |
| `Utils.retry` (amber, non-blocking) | unchanged | cross-references the
blocking sibling |
```scala
RetryUtil.withBackoff(description, maxAttempts, initialDelayMillis, onRetry, sleep)(operation)
```
The util owns everything the copies duplicated: attempt counting, the
doubling, what counts as transient (`NonFatal`), interrupt fail-fast
with the interrupt status restored, and the message wording.
`description` is a verb phrase ("connect to lake fs server"),
interpolated into all three messages, so LakeFS's give-up text is
byte-identical to before. Retries are reported through an `onRetry` hook
carrying a `RetryAttempt` (attempt, budget, delay, cause, and the
standard `message`), which callers log with **their own** logger — so a
LakeFS retry still logs under LakeFS, not under the util.
**Bug fixed on the way in.** LakeFS's `sleep` call sat inside the
`catch` block, so an interrupt raised *while waiting* escaped raw with
the interrupt flag left cleared — a `catch` cannot catch what its own
body throws:
```
Before (LakeFS): op fails -> catch -> sleep interrupted -> InterruptedException escapes, flag cleared
After (shared): op fails -> sleep interrupted -> interrupt restored + wrapped, fails fast
```
`FileService`'s copy already plugged that hole with an inner `try`; the
shared util keeps the better behavior for both.
**Why a new module rather than `common/workflow-core`**, which was
proposal 1 in apache#7095: `workflow-core` is not reachable from `Auth`,
`ConfigService`, or `AccessControlService`, so hosting it there would
have left future callers in those modules stuck writing their own loop —
the exact thing this PR is meant to stop. `common/config` is reachable
from everything and already carries
`org.apache.texera.amber.util.ConfigParserUtil`, but a retry helper
isn't configuration. A dependency-free module any other module may
depend on avoids both problems, at the cost of one `build.sbt` entry.
**Why the non-blocking variant stays in amber.** `Utils.retry` needs
`com.twitter:util-core`, declared only in `amber/build.sbt`. Moving it
down would put util-core on the classpath of every service that depends
on the new module and force LICENSE-binary resyncs, so the pair is:
blocking in `common/util`, non-blocking in `amber`, each documented in
terms of the other. The new module is dependency-free for the same
reason.
**One behavior change, deliberate.** Both replaced loops caught `case e:
Exception`; the util uses `NonFatal`, for parity with `Utils.retry`.
`NonFatal` is wider: non-fatal `Error`s — `AssertionError`,
`java.io.IOError`, `ServiceConfigurationError` — are now retried and
wrapped rather than propagating immediately. On the S3/LakeFS startup
paths that means such a failure costs the full backoff before surfacing.
`RetryUtilSpec` pins it in both directions (a non-fatal `Error` is
retried; a fatal throwable is not).
**Two retry sites deliberately left alone**, because converting them
changes behavior rather than just structure — tracked with the full
reasoning in apache#7124:
| Site | Why not now |
| --- | --- |
| `URLFetchUtil.getInputStreamFromURL` | Retries 5x with **no** delay
and returns `Option`. Adopting backoff adds up to ~3 s before a dead URL
gives up — arguably a fix (it is an unthrottled retry storm today), but
a behavior change in an operator path. |
| `PythonProxyClient` Flight connect loop | **Constant** delay, closes
the Flight client between attempts, and throws
`WorkflowRuntimeException` on give-up. Needs a delay-multiplier knob
plus a give-up type decision. |
### Any related issues, documentation, discussions?
Closes apache#7095
Follows apache#7088, which introduced the non-blocking variant; the
consolidation was split out of it to keep that fix narrow. Context: the
review discussion on apache#7103 (the LakeFS health-check backport) asked for
one util rather than a fourth copy.
### How was this PR tested?
`RetryUtilSpec` (new, in `common/util`, registered in `build.yml`'s
hand-maintained `jacoco` module list so it actually runs) is the union
of what the two hand-rolled loops were tested for, plus what neither
covered:
- returns the operation's value on first success without sleeping;
retries until success with a doubling progression; honors a custom base
delay; succeeds on the final permitted attempt (the `attempt >=
maxAttempts` boundary); gives up naming the description and preserving
the cause; `maxAttempts = 1` means no retry and no sleep;
- the `onRetry` hook fires once per wait with 1-based attempt numbers
and never after the last attempt, and its `message` wording is pinned;
- interrupt during the operation **and** interrupt during a backoff
sleep both restore the interrupt status and fail fast (the second is the
LakeFS hole above);
- a fatal throwable is not retried and passes through unwrapped, while a
non-fatal `Error` is retried (the `NonFatal`-vs-`Exception` widening
above).
At the call sites: `FileServiceSpec` keeps 8 tests that exercise the
delegation, its own 6/200 defaults, and the description in its messages
(the 4 that only re-tested util mechanics moved into `RetryUtilSpec`).
`LakeFSStorageClientSpec`'s 4 retry tests moved out with the loop; its
remaining 5 pass unchanged. amber's `UtilsSpec` and
`RegionExecutionManagerSpec` are untouched and still green.
```bash
sbt "Util/jacoco" "FileService/testOnly org.apache.texera.service.FileServiceSpec" "WorkflowCore/testOnly org.apache.texera.amber.core.storage.util.LakeFSStorageClientSpec"
```
```
[info] Tests: succeeded 11, failed 0, canceled 0, ignored 0, pending 0 (Util)
[info] Tests: succeeded 12, failed 0, canceled 0, ignored 0, pending 0 (FileService)
[info] Tests: succeeded 5, failed 0, canceled 0, ignored 0, pending 0 (WorkflowCore / LakeFS)
```
Lint plus test-source compiles for every module the new dependency edge
touches, and amber's two retry specs:
```bash
sbt "Util/scalafixAll --check" "Util/scalafmtCheckAll" "WorkflowCore/Test/compile" "WorkflowCore/scalafixAll --check" "FileService/Test/compile" "FileService/scalafixAll --check" "WorkflowExecutionService/Test/compile" "WorkflowExecutionService/scalafixAll --check" "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.common.UtilsSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec"
```
```
[success] all scalafix / scalafmt checks and Test/compile tasks
[info] Tests: succeeded 27, failed 0, canceled 0, ignored 0, pending 0
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
What changes were proposed in this PR?
Root cause.
RegionExecutionManagerretried region termination (EndWorker, thengracefulStop) up to 150 times at a flat 200 ms delay, so a region whose workers do not terminate held teardown for ~30 s before the failure surfaced. Killing a worker is deterministic --EndWorkerplusgracefulStopeither succeed, or something upstream is broken in a way retrying cannot fix (#6891, #6918). The long poll bought nothing except a quiet ~30 s stall that masked the real bug.IllegalStateExceptionnaming stuck workersThe retry itself lives in a util, not in the manager (per review):
Utils.retrykeeps itsattempts+baseBackoffTimeInMS-doubling contract but now schedules its waits on aTimerinstead of calling
Thread.sleep, so it can be used from an actor or coordinator thread -- ablocking sleep there stalls every other message queued on that thread, which is why this path could
not reuse the old version. An
onRetryhook keeps caller context in the log (the manager stillnames the region and the attempt); callers that do not care can omit it.
Utils.retry(attempts, baseBackoffTimeInMS, timer, onRetry) { operationReturningAFuture }The previous blocking
retryhad no production callers -- only its own tests -- so it was replacedrather than duplicated. A blocking front door can return as a thin sibling if a caller needs one.
RegionExecutionManageris now just a configuration of that util (4 attempts from a 200 ms base)plus its own give-up message. Two more copies of this loop live elsewhere --
LakeFSStorageClient.retryWithBackoffandFileService.awaitDependency-- but they sit incommon/workflow-coreandfile-service, neither of which can see amber'sUtils(andutil-coreis declared only in
amber/build.sbt). Folding those in means hosting the util lower in the modulegraph, tracked in #7095 rather than smuggled into this fix.
Log line, before and after:
This also shrinks -- but does not close -- the retry-on-
JavaTimer-thread race window in #6923, since there are far fewer hops onto that thread.Any related issues, documentation, discussions?
Closes #7087
Related: #7056 (the 150-attempt line appears in those hang logs), #6891, #6918, #6923. The bounded-retry loop this PR re-tunes was introduced in #5737.
How was this PR tested?
UtilsSpeccoversUtils.retrydirectly: first-attempt success waits not at all; the backoffdoubles (
200 -> 400 -> 800); waiting stops as soon as an attempt succeeds;onRetrysees the1-based failed-attempt number and the upcoming backoff; a synchronous throw from the by-name body
is retried like a failed
Futurerather than escaping; a fatal error is not retried in eithershape (it escapes on attempt 1, spending no backoff); and
attempts <= 1makes exactly one attemptwith no wait. The three cases that covered the removed blocking variant are gone with it.
RegionExecutionManagerSpeckeeps its termination-lifecycle cases and pins what the defaults meanend to end: the manager asks for
200 -> 400 -> 800 ms, makes 4EndWorkerattempts, and a regionthat terminates on attempt 2 spends only the first wait. Budget boundaries are still pinned
(success on the last permitted attempt, give-up at 1 attempt / 2 attempts with every stuck worker
named).
The backoff assertions are exact rather than timing-tolerant:
RecordingInlineTimerrecords eachrequested delay and runs the task inline, used inside
Time.withCurrentTimeFrozensowhen - Time.nowis precisely whatFuture.sleepasked for. The tests therefore assert the realschedule without waiting 1.4 s or flaking on slow CI. That needed a
killRetryTimerseam on theconstructor (default
new JavaTimer(true), as before).The two directly affected specs, with formatting and scalafix in the same invocation:
Whole scheduling package plus
UtilsSpec:sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.* org.apache.texera.amber.engine.common.UtilsSpec"(
DefaultCostEstimatorSpecaborts on this machine during suite construction, before any of itstests run, while building a CSV scan descriptor. Pre-existing and unrelated: with this PR's changes
reverted it aborts identically under the same command, 0 tests run.)
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)