You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/08/28 23:53:26 UTC
[PR] [WIP][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
sunchao opened a new pull request, #58402:
URL: https://github.com/apache/spark/pull/58402
### What changes were proposed in this pull request?
Jira creation is pending. This PR is a draft until the issue is created and linked.
Reject successful completions from invalidated barrier shuffle attempts before they can register map output, change pending-partition state, or notify a replacement task set that a partition is complete.
When a barrier shuffle stage fails and its outputs are cleared, call the existing `markAsRollingBack()` mechanism to invalidate results from that failed attempt and earlier attempts. Apply this at the two failure paths for the stage itself: ordinary barrier task failure and a fetch failure within a barrier shuffle stage. The existing `maxAttemptIdToIgnore` checks then suppress the stale completion before it affects recovery.
Keep valid replacement-attempt results and normal reuse of deterministic, non-barrier output unchanged. Do not mark an upstream barrier producer's running retry invalid when a downstream consumer reports a late fetch failure. No shuffle-manager or task-scheduler APIs change.
### Why are the changes needed?
Barrier recovery must relaunch every participant, even when the RDD is deterministic. The problematic ordering is:
1. One barrier task finishes, but its success reaches the driver late.
2. A peer fails, and Spark clears the entire barrier stage's map output.
3. The old success is accepted after invalidation.
Before resubmission, this can make the retry omit a partition and wait for a missing barrier participant. After resubmission but before launch, it can mark a required replacement partition complete and prevent the barrier task set from launching.
The existing handling for indeterminate rollback in [SPARK-45182](https://issues.apache.org/jira/browse/SPARK-45182), [SPARK-54556](https://issues.apache.org/jira/browse/SPARK-54556), and [SPARK-54956](https://issues.apache.org/jira/browse/SPARK-54956) does not establish invalidation for these deterministic barrier recovery paths.
### Does this PR introduce _any_ user-facing change?
Yes. Barrier shuffle retries retain all required participants when a successful completion arrives from an invalidated attempt. This avoids an unnecessary barrier timeout or a failure to launch in the affected orderings. There are no new public APIs or configuration options.
### How was this patch tested?
Compared public upstream baseline `024120dc960517021a887b51865220f544340362` with [patched commit daef30b0e585](https://github.com/sunchao/spark/commit/daef30b0e585c80786c2cc1f9304ce81204a190f), both reporting Spark `5.0.0-SNAPSHOT` with Scala `2.13.18`.
- Built core and its tests with the official Maven helper's `test-compile` goal on JDK 17. Ran the suites directly with ScalaTest: unpatched `DAGSchedulerSuite` had six new regression failures and 225 passes; patched `DAGSchedulerSuite` (231 tests) and `TaskSchedulerImplSuite` (121 tests) passed all 352 tests with no failures or aborted suites. These were direct ScalaTest runs, not Maven `test` executions.
- Ran six matched native local-mode scenarios per build on Java `21.0.12`, using `SortShuffleManager`, a deterministic two-partition RDD, real shuffle files, and downstream `collect()`. With a three-second barrier timeout, the unpatched before-resubmission case retried `2 -> 1` tasks and failed at a two-attempt limit, or recovered through `2 -> 1 -> 2` at a four-attempt limit. The patched cases both retried `2 -> 2` and returned the expected rows.
- In the blocked-slot case, the unpatched replacement became `successful=[true,false]`, with no running tasks, and did not launch before the 20-second job watchdog explicitly cancelled it. Slot release was about 17 seconds before that deadline. The patched replacement remained `successful=[false,false]`, launched both tasks after slot release, and succeeded. This is bounded no-progress evidence, not an observed permanent hang.
- The unblocked after-resubmission, no-delay barrier, and ordinary non-barrier controls succeeded on both builds. The unblocked case gates success on stage submission, not on both tasks having started. All six patched scenarios succeeded, all completed cases met their probe preconditions, and every successful run asserted `Seq((0, 0), (1, 1))`.
- Official Scalastyle passed for all three changed files with no errors or warnings.
The new scheduler regressions cover both failed-barrier-stage paths, before and after resubmission, valid replacement successes, ordinary deterministic output reuse, and a late consumer fetch failure that must not invalidate a producer's running retry. The native probe injects an ordinary task exception; native fetch-failure recovery and remote-cluster execution were not separately exercised.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex 0.151.0-alpha.7.1
This contribution is original work, licensed to the project under the Apache License, Version 2.0.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5460234306
Thanks, @dongjoon-hyun. I pushed the cleanup in [d7759d0](https://github.com/sunchao/spark/commit/d7759d0c4a7ab4460a6f403c8dd8a6e0c285a882):
- Restored the original `maxAttemptIdToIgnore` scaladoc and both original output-clearing comments.
- Renamed the barrier marker to `latestFailedBarrierAttemptId` and made the failure comment explain why late completions are ignored.
- Added `SPARK-59094:` to all eleven cases, fixed the singular/plural wording, and renamed the producer control to describe accepting successes after a late consumer fetch failure.
- Kept the direct completion-handler calls and their ordering comments: the tests intentionally avoid draining a timed resubmission event between the events under test.
- Shortened the PR description to the fix, scope, test coverage, and suite results.
Both suites were rerun: `DAGSchedulerSuite` 233/233 and `TaskSchedulerImplSuite` 121/121 passed.
For the producer-side follow-up, I confirmed a concrete ordering with a partially completed retry: accept one healthy producer-retry output, process the old consumer's second fetch failure, resubmit while the producer is still running, then complete its remaining task. An isolated scheduler characterization confirms that both the public baseline and this revision submit a one-task retry that still declares two barrier participants. This is scheduler-state evidence, not a native timeout reproduction.
The follow-up report is prepared, but its JIRA has not yet been filed; that is the remaining item.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5471451926
Thank you @sunchao and @dongjoon-hyun!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5459152468
Thanks for the fix! I went through the two patched failure paths and the surrounding rollback machinery. The core approach looks right to me.
Reusing `markAsRollingBack()` here is a good choice: since `ignoreOldTaskAttempts` compares with `>=`, only attempts up to the failed one are suppressed and the replacement attempt's results still land. And because the same flag also gates `notifyPartitionCompletion()`, it covers the "blocked slot" case (where the replacement task set gets a partition pre-marked complete and never launches), which a `findMissingPartitions()`-side fix would not. Both patched sites are behind the existing `latestInfo.attemptNumber() != task.stageAttemptId` guards, so they always apply to the latest attempt.
A few things I'd like your take on.
**1. Interaction with the `alreadyRollingBack` dedup guard**
`maxAttemptIdToIgnore` is also used as the dedup key for rollback:
```scala
val alreadyRollingBack = sms.maxAttemptIdToIgnore.contains(sms.latestInfo.attemptNumber())
if (sms.getNextAttemptId > 0 && !alreadyRollingBack) {
rollbackShuffleMapStage(sms, ...)
```
(`rollbackSucceedingStages`, and reached from `submitMissingTasks`.)
By setting `maxAttemptIdToIgnore = N` at barrier-failure time, a subsequent indeterminate rollback *while the stage is still at attempt N* now skips `rollbackShuffleMapStage()`. The `submitMissingTasks` rollback block runs before `makeNewStageAttempt()`, so `latestInfo.attemptNumber()` is still `N` there and it is reachable -- e.g. a barrier stage that is also statically indeterminate (`rdd.repartition(n).barrier()`).
What gets skipped is `newShuffleMergeState()`, which the comment right above that call describes as "creating a new shuffle merge state for the upcoming retry".
The practical impact today looks small: `canShuffleMergeBeEnabled()` disables push-based shuffle for barrier stages (`!rdd.isBarrier()`), so the only real effect is a missed `_shuffleMergeId` bump, which is inert there. But `maxAttemptIdToIgnore` now carries two meanings -- "invalidation marker" and "rollback dedup key" -- and this guard silently changes behavior as a result. Could you add a note about it? If SPARK-35547 ever enables push shuffle for barrier stages this becomes a real bug.
**2. The producer-side barrier path is left unprotected**
The other barrier branch in the `FetchFailed` handler is unchanged:
```scala
if (mapStage.rdd.isBarrier()) {
mapOutputTracker.unregisterAllMapAndMergeOutput(shuffleId)
```
A zombie/speculative task of the completed barrier producer whose success arrives after this `unregisterAll` re-registers that partition, and the following retry's `findMissingPartitions()` drops it -- the same class of barrier hang.
I agree with the reasoning in the PR description for not marking it (you must not invalidate a producer retry that is already running), and the regression test for it is a good addition. Since this leaves a pre-existing hole though, could you call it out explicitly as a remaining gap in the description, or file a follow-up JIRA? It also seems narrowable -- marking only when `mapStage` has not yet moved to a new attempt would cover the straggler case without touching a running retry.
**3. Minor**
- `Stage.markAsRollingBack()`'s `/** Mark the latest attempt as rollback */` could use the same update as the field scaladoc, since it now also covers barrier invalidation.
- The new comment replaced `// Mark all the map as broken in the map stage, to ensure retry all the tasks on resubmitted stage attempt.`, but that line explains the `unregisterAllMapAndMergeOutput()` call below it, while the new one explains `markAsRollingBack()`. I'd keep both.
Test coverage looks thorough -- before/after resubmission x 1-2 failed attempts, the `barrier=true/false` contrast confirming determinate non-barrier output is still reused, and the producer-retry regression. Using `scheduler.handleTaskCompletion()` directly to avoid the 200ms `scheduleResubmit()` timer event makes sense and there is precedent in the suite.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5459746930
Thanks for the detailed review, @dongjoon-hyun. I updated the comments and PR description.
1. The rollback interaction is fixed in [78f741b](https://github.com/sunchao/spark/commit/78f741b941d9e629d308f8e4d1ec66aba3f008b2). Barrier failures now use a separate `maxFailedBarrierAttemptId`, leaving `maxAttemptIdToIgnore` and rollback deduplication unchanged. The skipped cleanup can also matter today if a delayed shuffle-migration report restores an output before rollback cleanup. I added regressions for both ordinary task failure and `FetchFailed`: both pass on the public baseline, fail on the earlier shared-marker revision, and pass on the revised patch. `markAsRollingBack()` remains specific to rollback, so its existing method documentation still applies.
2. The producer-side `unregisterAllMapAndMergeOutput()` path is unchanged. The description now explicitly says that this PR does not protect already accepted retry outputs from a later downstream failure or add attempt validation to asynchronous migration updates. The producer-retry test checks that subsequent successes remain eligible, not that previously registered outputs survive clearing. For the particular completed-producer straggler scenario, I have not established a reachable ordering through normal scheduling: barrier speculation is disabled, normal stage completion processes all pending successes, duplicate terminal reports are dropped, and older failed or rolled-back attempts are already filtered. A concrete event ordering would help establish a separate regression before adding producer invalidation. This does not claim to resolve the broader producer-side recovery races.
3. I restored the explanation for clearing every map output at both failure sites, alongside the invalidation comment.
This follow-up changes only code comments and the PR description. Scalastyle and whitespace checks passed; scheduler and native tests were not rerun for the comment-only change. The description retains the earlier 354 scheduler tests and six native local scenarios, with their tested commit and coverage limits.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao closed pull request #58402: [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts
URL: https://github.com/apache/spark/pull/58402
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5481311707
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/7fa054b411af6d2d59b4e860e3d316af9e8594c0
- merged into branch-4.x https://github.com/apache/spark/commit/eb21775662994111c93f6a8d7adb1b4912249cac
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle attempts [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5481318515
Merged, thanks @dongjoon-hyun and @uros-b !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org