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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
