fix(amber): bound region termination retry instead of looping forever - #5737
Merged
xuang7 merged 16 commits intoJun 27, 2026
Conversation
…or user to see and fix
…nation-bounded-retry
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 435 | 0.266 | 23,215/30,690/30,690 us | 🔴 +8.9% / 🔴 +97.0% |
| 🟢 | bs=100 sw=10 sl=64 | 959 | 0.585 | 103,890/120,058/120,058 us | 🟢 -24.4% / 🔴 +9.5% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,114 | 0.68 | 902,990/949,073/949,073 us | ⚪ within ±5% / 🟢 -12.3% |
Baseline details
Latest main 7a38b6c from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 435 tuples/sec | 465 tuples/sec | 758.88 tuples/sec | -6.5% | -42.7% |
| bs=10 sw=10 sl=64 | MB/s | 0.266 MB/s | 0.284 MB/s | 0.463 MB/s | -6.3% | -42.6% |
| bs=10 sw=10 sl=64 | p50 | 23,215 us | 21,310 us | 12,965 us | +8.9% | +79.1% |
| bs=10 sw=10 sl=64 | p95 | 30,690 us | 30,290 us | 15,578 us | +1.3% | +97.0% |
| bs=10 sw=10 sl=64 | p99 | 30,690 us | 30,290 us | 18,378 us | +1.3% | +67.0% |
| bs=100 sw=10 sl=64 | throughput | 959 tuples/sec | 939 tuples/sec | 968.9 tuples/sec | +2.1% | -1.0% |
| bs=100 sw=10 sl=64 | MB/s | 0.585 MB/s | 0.573 MB/s | 0.591 MB/s | +2.1% | -1.1% |
| bs=100 sw=10 sl=64 | p50 | 103,890 us | 104,630 us | 102,767 us | -0.7% | +1.1% |
| bs=100 sw=10 sl=64 | p95 | 120,058 us | 158,773 us | 109,629 us | -24.4% | +9.5% |
| bs=100 sw=10 sl=64 | p99 | 120,058 us | 158,773 us | 118,129 us | -24.4% | +1.6% |
| bs=1000 sw=10 sl=64 | throughput | 1,114 tuples/sec | 1,113 tuples/sec | 997.01 tuples/sec | +0.1% | +11.7% |
| bs=1000 sw=10 sl=64 | MB/s | 0.68 MB/s | 0.679 MB/s | 0.609 MB/s | +0.1% | +11.7% |
| bs=1000 sw=10 sl=64 | p50 | 902,990 us | 895,733 us | 1,009,306 us | +0.8% | -10.5% |
| bs=1000 sw=10 sl=64 | p95 | 949,073 us | 955,612 us | 1,051,088 us | -0.7% | -9.7% |
| bs=1000 sw=10 sl=64 | p99 | 949,073 us | 955,612 us | 1,082,535 us | -0.7% | -12.3% |
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,459.28,200,128000,435,0.266,23215.27,30689.55,30689.55
1,100,10,64,20,2086.08,2000,1280000,959,0.585,103889.63,120057.57,120057.57
2,1000,10,64,20,17955.11,20000,12800000,1114,0.680,902989.93,949073.45,949073.45…7Ball/texera into fix/region-termination-bounded-retry
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #5737 +/- ##
============================================
- Coverage 54.89% 54.82% -0.08%
+ Complexity 2962 2958 -4
============================================
Files 1117 1115 -2
Lines 43133 43037 -96
Branches 4648 4636 -12
============================================
- Hits 23680 23597 -83
+ Misses 18064 18053 -11
+ Partials 1389 1387 -2
*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:
|
Contributor
Automated Reviewer SuggestionsBased on the
|
Ma77Ball
marked this pull request as ready for review
June 23, 2026 10:12
Contributor
Author
|
/request-review @aglinxinyuan |
Ma77Ball
force-pushed
the
fix/region-termination-bounded-retry
branch
from
June 23, 2026 10:15
f16b303 to
7ec4e83
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR updates Amber’s region termination behavior so that RegionExecutionCoordinator.terminateWorkersWithRetry no longer retries EndWorker indefinitely. Instead, it enforces a bounded retry budget (production defaults ≈30s) and fails fast with a descriptive error that includes the region id and affected worker ids, improving diagnosability of CI flakes like the DataProcessingSpec timeout described in #5614.
Changes:
- Add bounded termination retry logic with configurable
maxTerminationAttemptsandkillRetryDelay, plus descriptive failure on exhaustion. - Add/extend scheduler test support utilities for multi-worker regions in tests.
- Add unit tests covering retry-budget exhaustion, boundary (off-by-one) behavior, multi-worker aggregation, and default values.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala | Introduces bounded termination retry parameters + defaults; fails termination with a descriptive exception after budget exhaustion. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinatorSpec.scala | Adds focused tests for give-up behavior, boundary conditions, multi-worker error reporting, and default budget values. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala | Adds helpers to create/seed regions with multiple workers to support the new tests. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
aglinxinyuan
approved these changes
Jun 24, 2026
aglinxinyuan
left a comment
aglinxinyuan
left a comment
Contributor
There was a problem hiding this comment.
LGTM — bounding the retry is the right call. I traced the failure path to confirm it does what the description says: on budget exhaustion the failed terminationFuture skips the setPhase(Completed) flatMap, propagates through coordinateRegionExecutors' Future.collect, and surfaces as a FatalError in PortCompletedHandler (hits the case other branch, so relatedWorkerId = None — but the message itself carries the worker ids). Defaults are tunable and the only other constructor call site uses them, so the blast radius is minimal. Test coverage is good too — the "succeeds on the last permitted attempt" case nicely pins the attempt >= maxTerminationAttempts off-by-one.
One thing to confirm before merge — the budget is ~30s, but the stalls in #5614 showed ~60s silent windows. In production (no test timeout) a worker that would have drained late now gets the workflow failed at 30s instead of eventually completing. Are we comfortable that 30s is enough headroom for a slow-but-recoverable drain? Fine with it either way since it's a constructor param — just want it to be a deliberate choice rather than fall out of the 150 * 200ms arithmetic.
Couple of optional nits:
- The give-up message lists every region worker as "still not terminated," but
terminateWorkersfails the whole batch viaFuture.collectif any singleEndWorkerfails — so a worker that terminated fine still gets named. Something like "region workers" would read more accurately. - The companion-object comment says "Max EndWorker retries," but
maxTerminationAttemptscounts attempts (1 initial + 149 retries = 150) — minor wording.
Agree with the scope note that this is a fail-fast/diagnosability change, not a fix for the underlying termination flake (the Iceberg-commit contention during teardown is the real culprit) — but it's a strictly better failure mode than a silent hang, so no objection to merging once the budget question is settled.
Contributor
Author
aglinxinyuan
pushed a commit
that referenced
this pull request
Jun 27, 2026
…#5958) ### What changes were proposed in this PR? - Backport of #5737 to `release/v1.2`: bound the EndWorker termination retry loop with a configurable budget (`maxTerminationAttempts`, default 150) and per-attempt delay (`killRetryDelay`, default 200ms, ~30s total) instead of retrying forever. - When the budget is exhausted, the region fails loudly with an `IllegalStateException` that names every still-running worker and chains the underlying cause, so a stuck region surfaces to the user instead of hanging silently. - Adds `RegionExecutionCoordinator.DefaultMaxTerminationAttempts` / `DefaultKillRetryDelay`, threads the two new constructor params through, and extends the test fixtures with give-up and boundary coverage. ### Any related issues, documentation, discussions? Backport of #5737 to release/v1.2. ### How was this PR tested? - Run `sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionCoordinatorSpec"` (JDK 17); expect all 7 tests to pass (2 existing retry tests plus 5 new give-up/boundary tests). - Give-up path: a worker that never drains fails after the budget with "could not be terminated after N attempts" naming the stuck worker(s); the single-attempt, last-attempt-success, multi-worker aggregation, and default-budget cases are each pinned by a new spec. ### Was this PR authored or co-authored using generative AI tooling? Co-authored with Claude Opus 4.8 in compliance with ASF
This was referenced Jul 25, 2026
Comment on lines
+68
to
+74
|
|
||
| // Max EndWorker retries before termination fails. ~30s at DefaultKillRetryDelay (200ms). | ||
| private[scheduling] val DefaultMaxTerminationAttempts: Int = 150 | ||
|
|
||
| private[scheduling] val DefaultKillRetryDelay: TwitterDuration = | ||
| TwitterDuration.fromMilliseconds(200) | ||
| } |
Contributor
There was a problem hiding this comment.
this number is too large! why does it need to retry for 150 times to kill a worker? this uptime to 30s per worker. in integration tests this will easily rocket up.
Example:
https://github.com/apache/texera/actions/runs/30479437487/job/90673007976?pr=7042
Can we change this to 3 retries max?
At the end of day, we should fix the flakiness, there shouldn't be retries.
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? **Root cause.** `RegionExecutionManager` retried region termination (`EndWorker`, then `gracefulStop`) 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 -- `EndWorker` plus `gracefulStop` either succeed, or something upstream is broken in a way retrying cannot fix (apache#6891, apache#6918). The long poll bought nothing except a quiet ~30 s stall that masked the real bug. ``` Before: EndWorker fails -> 149 retries x 200 ms -> ~30 s of silence -> loud failure After: EndWorker fails -> 3 retries (200/400/800 ms) -> ~1.4 s -> loud failure ``` | | Before | After | | --- | --- | --- | | Attempts | 150 | 4 (1 + 3 retries) | | Delay between attempts | flat 200 ms | 200 -> 400 -> 800 ms | | Worst-case wait per region | ~30 s | ~1.4 s | | Give-up behavior | `IllegalStateException` naming stuck workers | unchanged | **The retry itself lives in a util, not in the manager** (per review): `Utils.retry` keeps its `attempts` + `baseBackoffTimeInMS`-doubling contract but now schedules its waits on a `Timer` instead of calling `Thread.sleep`, so it can be used from an actor or coordinator thread -- a blocking sleep there stalls every other message queued on that thread, which is why this path could not reuse the old version. An `onRetry` hook keeps caller context in the log (the manager still names the region and the attempt); callers that do not care can omit it. ```scala Utils.retry(attempts, baseBackoffTimeInMS, timer, onRetry) { operationReturningAFuture } ``` The previous blocking `retry` had no production callers -- only its own tests -- so it was replaced rather than duplicated. A blocking front door can return as a thin sibling if a caller needs one. `RegionExecutionManager` is 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.retryWithBackoff` and `FileService.awaitDependency` -- but they sit in `common/workflow-core` and `file-service`, neither of which can see amber's `Utils` (and `util-core` is declared only in `amber/build.sbt`). Folding those in means hosting the util lower in the module graph, tracked in apache#7095 rather than smuggled into this fix. Log line, before and after: ``` - Failed to terminate region 1 on attempt 1 of 150. Retrying in 200 ms. + Failed to terminate region 1 on attempt 1 of 4. Retrying in 200 ms. + Failed to terminate region 1 on attempt 2 of 4. Retrying in 400 ms. + Failed to terminate region 1 on attempt 3 of 4. Retrying in 800 ms. + Region 1 could not be terminated after 4 attempts; giving up. Workers still not terminated: ... ``` This also shrinks -- but does not close -- the retry-on-`JavaTimer`-thread race window in apache#6923, since there are far fewer hops onto that thread. ### Any related issues, documentation, discussions? Closes apache#7087 Related: apache#7056 (the 150-attempt line appears in those hang logs), apache#6891, apache#6918, apache#6923. The bounded-retry loop this PR re-tunes was introduced in apache#5737. ### How was this PR tested? `UtilsSpec` covers `Utils.retry` directly: first-attempt success waits not at all; the backoff doubles (`200 -> 400 -> 800`); waiting stops as soon as an attempt succeeds; `onRetry` sees the 1-based failed-attempt number and the upcoming backoff; a *synchronous* throw from the by-name body is retried like a failed `Future` rather than escaping; a *fatal* error is not retried in either shape (it escapes on attempt 1, spending no backoff); and `attempts <= 1` makes exactly one attempt with no wait. The three cases that covered the removed blocking variant are gone with it. `RegionExecutionManagerSpec` keeps its termination-lifecycle cases and pins what the defaults mean end to end: the manager asks for `200 -> 400 -> 800 ms`, makes 4 `EndWorker` attempts, and a region that 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: `RecordingInlineTimer` records each requested delay and runs the task inline, used inside `Time.withCurrentTimeFrozen` so `when - Time.now` is precisely what `Future.sleep` asked for. The tests therefore assert the real schedule without waiting 1.4 s or flaking on slow CI. That needed a `killRetryTimer` seam on the constructor (default `new JavaTimer(true)`, as before). The two directly affected specs, with formatting and scalafix in the same invocation: ```bash sbt "WorkflowExecutionService/scalafmtAll" "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.common.UtilsSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec" "WorkflowExecutionService/scalafixAll --check" ``` ``` [info] Tests: succeeded 27, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. ``` Whole scheduling package plus `UtilsSpec`: ```bash sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.* org.apache.texera.amber.engine.common.UtilsSpec" ``` ``` [info] Suites: completed 13, aborted 1 [info] Tests: succeeded 145, failed 0, canceled 0, ignored 0, pending 0 ``` (`DefaultCostEstimatorSpec` aborts on this machine during suite construction, before any of its tests 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)
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
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?
RegionExecutionCoordinator.terminateWorkersWithRetryretried worker termination with no limit, so a worker that never finished draining (itsEndWorkerkept failing) left the region's termination future unresolved and the workflow never reached COMPLETED, surfacing as an opaque ~1 minute timeout inDataProcessingSpecwith no indication of the stuck region or workers.maxTerminationAttemptsbudget (default 150, about 30s at the existing 200ms delay); on exhaustion the termination future fails with a descriptive error naming the region and the still-unterminated workers, instead of retrying indefinitely.maxTerminationAttemptsandkillRetryDelayconstructor parameters with production defaults so the behavior is unit-testable without long waits.Any related issues, documentation, discussions?
Closes: #5614
How was this PR tested?
sbt "WorkflowExecutionService/testOnly *RegionExecutionCoordinatorSpec"(under JDK 17); expect 3 tests succeeded, 0 failed.give up with a descriptive error once the EndWorker retry budget is exhausted: forcesEndWorkerto always fail, then asserts the completion future fails withIllegalStateExceptioncontaining "could not be terminated after 3 attempts", the coordinator is not marked completed, exactly 3EndWorkercalls were made, and the worker actor ref is retained.Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8 in compliance with ASF