jerrypeng opened a new pull request, #59499:
URL: https://github.com/apache/spark/pull/59499
### What changes were proposed in this pull request?
A pipelined-group task set caps attempts at one and counts every failure,
even reasons that are normally not counted, so that a member failure fails the
whole group. `TaskSetManager.handleFailedTask` exempted `TaskKilled` and
`TaskCommitDenied` from that rule. Neither counts for a regular task set, so a
killed or commit-denied pipelined task was re-queued and relaunched in the same
stage attempt.
This PR counts every failure of a live pipelined attempt
(`reason.countTowardsTaskFailures || taskSet.isPipelined`). The first one
aborts the task set, and the existing abort path fails the job and the rest of
its group. Kills that the scheduler sends while failing, cancelling, or
finishing a job are unaffected: `killAllTaskAttempts` suspends the task set
before those kills are reported, and a suspended (zombie) task set does not
count failures. Jobs without a pipelined shuffle are unchanged.
### Why are the changes needed?
A pipelined member's peers have already exchanged part of its transient
stream, so retrying the task within the same stage attempt is unsound. The
exemption was meant for losing duplicate attempts and commit races, but a
pipelined task never has a second attempt: a pipelined job rejects speculation,
and with this change a failure aborts the task set instead of retrying the task.
With the in-process channel transport, in local mode, on master:
- Killing a producer task with `SparkContext.killTaskAttempt` after it had
queued two batches relaunched it in the same stage attempt. The job succeeded,
with the queued rows delivered twice.
- Killing a consumer task after it had read some rows relaunched it. The job
hung: the killed attempt had already drained the queue it was reading.
### Does this PR introduce _any_ user-facing change?
Yes, for jobs with a pipelined shuffle dependency. A killed task (for
example through `SparkContext.killTaskAttempt`) or a denied commit in such a
job now fails the job, as any other failure of a pipelined task does. Before,
the task was retried in place, which could duplicate rows or hang the job. Jobs
without a pipelined shuffle dependency are unchanged.
### How was this patch tested?
- Replaced the two `TaskSetManagerSuite` tests that asserted the exemption
(neither established a winning duplicate attempt). The new tests check that a
kill and a denied commit of a running pipelined task abort the task set, make
it a zombie, and offer no replacement task.
- New `TaskSetManagerSuite` tests:
- Kills and denied commits reported after a pipelined task set was
aborted, or after `suspend()` (what `killAllTaskAttempts` does), neither abort
it again nor relaunch a task, and a repeated report for a finished attempt is
ignored.
- A regular task set with a one-failure limit still retries a task after a
kill or a denied commit.
- New `PipelinedChannelShuffleSuite` tests, in local mode, each killing a
task mid-stream with `SparkContext.killTaskAttempt`:
- Kill the first attempt of a producer task after it has queued two
batches. The consumer task is killed with the group, and both stages end as
failed.
- Kill the consumer task (a result-stage member) after it has read two
batches. Its stage ends as failed.
- In both, the job fails once, after one failure, with the kill reason; no
task or stage is retried; every started task ends; no job or channel queue is
left behind; and a fresh rerun returns the correct rows.
- Without this change, the kill and denied-commit tests fail, the producer
test's job succeeds, and the consumer test's job hangs until the test times
out. Mutants that exempt either reason, drop the zombie check, or count
failures for every task set each fail at least one new test.
- Ran locally with Java 17, recompiling the changed files with Spark's
scalac flags (warnings as errors) against an sbt build of the base commit, with
snappy-java pinned to 1.1.10.11 because the local environment could not resolve
1.1.10.12. These suites all passed, 907 tests: `TaskSetManagerSuite`,
`TaskSchedulerImplSuite`, `DAGSchedulerSuite`,
`DAGSchedulerAbortStageOffSuite`, `PipelinedChannelShuffleSuite`,
`StreamingShuffleSuite`, `PipelinedShuffleRoutingSuite`,
`ShuffleDependencySuite`, `OutputCommitCoordinatorSuite`,
`BarrierStageOnSubmittedSuite`, and `JobCancellationSuite`. So did the
cancellation tests in `SparkContextSuite`. Scalastyle reported no errors or
warnings on the changed files. To run the new tests:
```
build/sbt 'core/testOnly org.apache.spark.scheduler.TaskSetManagerSuite
org.apache.spark.shuffle.local.pipelined.PipelinedChannelShuffleSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (2.1.299, Claude Opus 5.5)
--
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]
