ivoson opened a new pull request, #59793:
URL: https://github.com/apache/spark/pull/59793
### What changes were proposed in this pull request?
This PR initializes `Stage.maxChecksumMismatchedId` to `-1` instead of
`nextAttemptId`, which is always `0` when a stage is constructed.
When a shuffle checksum mismatch is detected while registering map output,
`DAGScheduler` rolls back the succeeding stages only if the following guard
passes:
```scala
if (shuffleStage.maxChecksumMismatchedId < smt.stageAttemptId) {
shuffleStage.maxChecksumMismatchedId = smt.stageAttemptId
if (shuffleStage.shuffleDep.checksumMismatchQueryLevelRollbackEnabled) {
rollbackSucceedingStagesForQuery(shuffleStage)
} else {
rollbackSucceedingStages(shuffleStage)
}
}
```
With the marker starting at `0`, the guard can never pass for a task from
stage attempt `0`. With `-1`, the first checksum mismatch in attempt `0` goes
through the existing rollback/abort handling. Later mismatches in the same
attempt are still deduplicated, and the behavior for attempts `1` and later is
unchanged.
### Why are the changes needed?
A checksum mismatch can be detected by a task from stage attempt `0` in two
cases, and both were silently accepted:
1. **A shuffle map stage is recreated for an existing shuffle.** When a job
finishes, its stages are cleaned up, but `MapOutputTracker` keeps the shuffle's
map statuses and their checksums. If some map output is later lost and a
subsequent job that depends on the same shuffle recomputes the missing
partitions, the new `ShuffleMapStage` starts at attempt `0`. If the recomputed
output differs from the previous output, the mismatch is detected in attempt
`0`.
2. **A late task from attempt `0` replaces the output of a later attempt.**
When attempt `0` of a non-barrier shuffle map stage fails with a `FetchFailed`,
its other running tasks are not killed. If one of them finishes after attempt
`1` has recomputed the same partition, it replaces attempt `1`'s registered map
output. A changed checksum here was ignored, while the reverse order (a late
task from attempt `1` replacing attempt `0`'s output) already triggered
rollback.
In both cases, consumers that already read the previous output keep results
computed from it, which can lead to incorrect query results.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UTs added.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (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]
