[SPARK-58398][CORE] Group-atomic failure and fail-fast rejection for pipelined-shuffle stage groups

### What changes were proposed in this pull request?

This PR combines the two failure-handling layers of the pipelined-shuffle work into a single
reviewable change, on top of the now-merged concurrent scheduling of a `PipelinedShuffleDependency`
group (SPARK-58263 / #57341): the **group-atomic failure model** and the **fail-fast rejection of
unsupported idioms** that together make co-scheduling safe under failure. (It builds on two merged
predecessors -- the `PipelinedShuffleDependency` definition and type-based routing from
SPARK-58185 / #57286, and the concurrent scheduling from SPARK-58263 / #57341 -- and targets
`master` directly; its net-new content is the failure and fail-fast layers described below.)

A pipelined shuffle is transient and incrementally read: its output is streamed to a co-scheduled
consumer and is never materialized as durable, re-readable map output. That has two consequences the
scheduler must enforce.

**Group-atomic failure (a pipelined group succeeds or fails as a whole).** A transient shuffle
cannot be recomputed in isolation and its members run concurrently, so the stock "resubmit one
stage" recovery does not apply -- any member failure must fail the whole group, which the caller
then reruns (a streaming query restarts the batch).

- **Member task failure.** A pipelined group member's `TaskSet` is tagged `isPipelined` and pinned
  to `maxTaskFailures = 1`, so the first task failure aborts the set instead of retrying in place,
  which routes to a whole-group abort.
- **Executor loss.** Losing an executor running a member task is force-counted as a member failure
  (except benign `TaskKilled` / `TaskCommitDenied`), aborting the group.
- **FetchFailed on a member.** Handled by a dedicated branch that aborts the whole group rather than
  resubmitting the map stage in isolation (a lone-stage resubmit of a transient shuffle would
  deadlock the group). The failed executor's *regular* outputs are still unregistered from the
  `MapOutputTracker`, exactly as the base handler does, for the benefit of other jobs.
- **No transient-producer resubmit.** A pipelined `ShuffleMapStage` records its completed partitions
  locally and monotonically (never in the `MapOutputTracker`) and never flips back to unavailable, so
  losing an executor that held an already-consumed pipelined output cannot make the scheduler
  resubmit the producer -- which would otherwise hang the producer's streaming writer waiting on
  termination acks from reducers that already finished. The `TaskSetManager` "Resubmitted"
  re-enqueue loop likewise excludes pipelined sets.
- **Cross-job / cross-time reuse rejected.** A transient shuffle has no retained output for a second
  job to read, so binding a pipelined producer stage to a second job fails fast.

**Fail-fast on unsupported idioms (spec S9).** A pipelined group is rejected up front, before any
stage is created (so a rejection leaves no partial scheduler state), when it uses an idiom v1 cannot
support: a producer feeding **more than one consumer** (1:N fan-out needs multicast, deferred); a
**barrier**, **statically-indeterminate**, **checksum-mismatch-retry**, or **push-merge** producer;
a **reliable RDD checkpoint** anywhere in a member's within-stage chain (it reintroduces cross-time
reuse of a transient edge); or **members with differing resource profiles** (the gang slot check
compares one demand against one profile's capacity, so v1 requires a single-profile group). These
throw a typed `PipelinedShuffleUnsupportedException` (carrying the `PIPELINED_SHUFFLE_UNSUPPORTED`
error class), which `handleJobSubmitted` matches by type.

Main changes:

- `DAGScheduler.scala` -- the FetchFailed group-abort branch, the no-resubmit handling of a
  pipelined `ShuffleMapStage`, cross-job reuse rejection, and `checkPipelinedGroupsSupportedInRDDGraph`
  / `checkPipelinedProducerSupported` fail-fast (typed exception).
- `TaskSetManager.scala` -- `maxTaskFailures = 1` and force-counted executor loss for a pipelined
  set; exclusion from the "Resubmitted" re-enqueue loop.
- `ShuffleMapStage.scala` -- monotonic local availability for a pipelined shuffle.
- `PIPELINED_SHUFFLE_UNSUPPORTED` and `PIPELINED_SHUFFLE_CROSS_JOB_REUSE` error conditions.

### Why are the changes needed?

Co-scheduling a pipelined group (#57341) is only safe if failure is handled at the granularity of
the whole group: because the shuffle is transient and once-through, the stock per-stage resubmit
recovery would either deadlock the group or hang a producer's streaming writer. This PR makes any
member failure fail the group atomically (so the caller reruns it) and rejects up front the idioms
whose recovery/semantics are incompatible with a transient, concurrently-read shuffle -- turning
what would be a hang or a silently-wrong schedule into a clear, immediate failure. The two layers
are combined into one PR because they are inseparable in review: the fail-fast rejections define
exactly which group shapes the failure model must then handle, and both are gated on the same
`PipelinedShuffleDependency` type.

### Does this PR introduce _any_ user-facing change?

No. All new behavior is gated on a job using a `PipelinedShuffleDependency`, which nothing constructs
yet, so every existing job is scheduled and recovers exactly as before. The new error conditions
(`PIPELINED_SHUFFLE_UNSUPPORTED`, `PIPELINED_SHUFFLE_CROSS_JOB_REUSE`) can only surface for a job
that uses a pipelined dependency.

### How was this patch tested?

New unit tests in `DAGSchedulerSuite` and `TaskSetManagerSuite` cover:

- group-atomic failure: `maxTaskFailures = 1` aborting a member set on the first failure; executor
  loss force-counted (and benign `TaskKilled` / `TaskCommitDenied` not force-counted); a member
  FetchFailed aborting the whole group rather than resubmitting a single stage;
- no transient-producer resubmit: a post-executor-loss straggler success not resubmitting the
  producer; a completed pipelined producer's availability surviving executor loss; the "Resubmitted"
  loop excluding a pipelined set (including the partial-producer-on-decommission case);
- cross-job reuse rejected; a group-atomic rerun resetting per-partition commit authorization;
- fail-fast idioms: fan-out, barrier / indeterminate / checksum / push-merge producer, a reliable
  checkpoint in a producer's or a consumer's chain (including downstream in the consumer stage), and
  a mixed-resource-profile group -- each rejected up front; and that regular-shuffle idioms are NOT
  rejected (inertness of the fail-fast for a job with no pipelined dependency).

The full `DAGSchedulerSuite` and `TaskSetManagerSuite` pass.

### Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Code (Opus 4.8)

Closes #57361 from jerrypeng/stack/pipelined-shuffle-pr6-failfast.

Authored-by: Boyang Jerry Peng <jerry.peng@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
8 files changed
tree: 898959e04e537e149c50459160d6265bdfd5f7cf
  1. .github/
  2. .mvn/
  3. assembly/
  4. bin/
  5. binder/
  6. build/
  7. common/
  8. conf/
  9. connector/
  10. core/
  11. data/
  12. dev/
  13. docs/
  14. examples/
  15. graphx/
  16. hadoop-cloud/
  17. launcher/
  18. licenses/
  19. licenses-binary/
  20. mllib/
  21. mllib-local/
  22. project/
  23. python/
  24. R/
  25. repl/
  26. resource-managers/
  27. sbin/
  28. sql/
  29. streaming/
  30. tools/
  31. udf/
  32. ui-test/
  33. .asf.yaml
  34. .gitattributes
  35. .gitignore
  36. .nojekyll
  37. .pre-commit-config.yaml
  38. .sbtopts
  39. AGENTS.md
  40. CONTRIBUTING.md
  41. LICENSE
  42. LICENSE-binary
  43. NOTICE
  44. NOTICE-binary
  45. pom.xml
  46. pyproject.toml
  47. README.md
  48. scalastyle-config.xml
  49. SECURITY.md
README.md

Apache Spark

Spark is a unified analytics engine for large-scale data processing. It provides high-level APIs in Scala, Java, Python, and R (Deprecated), and an optimized engine that supports general computation graphs for data analysis. It also supports a rich set of higher-level tools including Spark SQL for SQL and DataFrames, pandas API on Spark for pandas workloads, MLlib for machine learning, GraphX for graph processing, and Structured Streaming for stream processing.

License Maven Central Java GitHub Actions Build PySpark Coverage PyPI Downloads

Online Documentation

You can find the latest Spark documentation, including a programming guide, on the project web page. This README file only contains basic setup instructions.

Build Pipeline Status

BranchStatus
masterGitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.xGitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.2GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.1GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.0GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-3.5GitHub Actions Build
GitHub Actions Build
GitHub Actions Build

Building Spark

Spark is built using Apache Maven. To build Spark and its example programs, run:

./build/mvn -DskipTests clean package

(You do not need to do this if you downloaded a pre-built package.)

More detailed documentation is available from the project site, at “Building Spark”.

For general development tips, including info on developing Spark using an IDE, see “Useful Developer Tools”.

Interactive Scala Shell

The easiest way to start using Spark is through the Scala shell:

./bin/spark-shell

Try the following command, which should return 1,000,000,000:

scala> spark.range(1000 * 1000 * 1000).count()

Interactive Python Shell

Alternatively, if you prefer Python, you can use the Python shell:

./bin/pyspark

And run the following command, which should also return 1,000,000,000:

>>> spark.range(1000 * 1000 * 1000).count()

Example Programs

Spark also comes with several sample programs in the examples directory. To run one of them, use ./bin/run-example <class> [params]. For example:

./bin/run-example SparkPi

will run the Pi example locally.

You can set the MASTER environment variable when running examples to submit examples to a cluster. This can be spark:// URL, “yarn” to run on YARN, and “local” to run locally with one thread, or “local[N]” to run locally with N threads. You can also use an abbreviated class name if the class is in the examples package. For instance:

MASTER=spark://host:7077 ./bin/run-example SparkPi

Many of the example programs print usage help if no params are given.

Running Tests

Testing first requires building Spark. Once Spark is built, tests can be run using:

./dev/run-tests

Please see the guidance on how to run tests for a module, or individual tests.

There is also a Kubernetes integration test, see resource-managers/kubernetes/integration-tests/README.md

A Note About Hadoop Versions

Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported storage systems. Because the protocols have changed in different versions of Hadoop, you must build Spark against the same version that your cluster runs.

Please refer to the build documentation at “Specifying the Hadoop Version and Enabling YARN” for detailed guidance on building for a particular distribution of Hadoop, including building for particular Hive and Hive Thriftserver distributions.

Configuration

Please refer to the Configuration Guide in the online documentation for an overview on how to configure Spark.

Contributing

Please review the Contribution to Spark guide for information on how to get started contributing to the project.