wang-haihua opened a new pull request, #58062:
URL: https://github.com/apache/spark/pull/58062
### What changes were proposed in this pull request?
This PR adds an integrity checksum to `MapStatus` that covers the *set of
non-empty
partition indices*, and verifies it wherever a `MapStatus` is consumed to
plan shuffle
block fetches.
- A new `MapStatusChecksum` utility computes a checksum (CRC32 or Adler32)
over the
indices of non-empty partitions in a `partitionLengths` array. It returns
`None` when
there are no non-empty partitions.
- `MapStatus` (both `CompressedMapStatus` and `HighlyCompressedMapStatus`)
gains a new
`nonEmptyChecksum: Option[Int]` field, computed at construction time when
enabled, and
carried through Java (`Externalizable`) serialization via a presence flag
followed by
the value, matching the existing pattern used for other optional fields.
- `MapOutputTracker.verifyChecksumOrFail` recomputes the checksum from the
`MapStatus`
actually consumed by a reducer and compares it against the stored value. A
mismatch
throws a `FetchFailedException`, reusing Spark's existing fetch-failure /
stage-retry
recovery path instead of introducing a new failure mode.
- Two new internal, off-by-default configs:
- `spark.shuffle.mapStatus.checksum.enabled` (default `false`)
- `spark.shuffle.mapStatus.checksum.algorithm` (default `CRC32`, also
supports `ADLER32`)
The checksum is computed over the non-empty/empty *boundary* of each
partition rather
than the sizes themselves, because sizes are already lossily
compressed/estimated by
`CompressedMapStatus` / `HighlyCompressedMapStatus` and can legitimately
differ in
representation while still being correct. Only the non-empty/empty boundary
matters for
the silent-data-loss failure mode this PR targets.
### Why are the changes needed?
`MapStatus` is the most frequently accessed piece of shuffle metadata: every
mapper
produces one, the driver holds one per map task, and every reducer consults
it to decide
which blocks to fetch. If a `MapStatus`'s `partitionLengths` (or its
compressed
representation) is corrupted in memory or in transit between the mapper and
a reducer --
e.g. due to a transient hardware fault, a JVM/memory corruption, or a
serialization bug
-- a partition that actually has data on disk can appear to have size 0.
Because block
fetchers are intentionally designed to skip zero-size blocks
(`MapStatus.getSizeForBlock(reduceId) == 0`), the reducer silently skips
fetching that
block. The on-disk data is never read and is effectively lost, with no
exception, no
failed stage, and no log signal anywhere in the pipeline.
We hit this in production: a job produced fewer output rows than an
otherwise-equivalent
rerun, with no errors in the driver or executor logs. Comparing Spark UI SQL
metrics for
the two runs showed shuffle "records written" and "records read" diverging
by a large
margin at an Exchange node in the faulty run. Instrumenting `MapStatus` at
several points
along its mapper -> driver -> reducer path showed a handful of non-empty
partition
entries had been flipped to zero somewhere along that path, while the
corresponding
shuffle data files on disk were complete and correct. The root cause traced
back to a
hardware fault on one worker node that silently corrupted the `MapStatus`
object held in
the driver's memory. Full writeup: SPARK-58830.
Existing safeguards don't catch this class of failure:
- SPARK-35276 added a checksum over shuffle *data* files, so disk/IO-level
corruption of
shuffle bytes is detectable when a reducer actually reads them. But when
the corruption
is in the *metadata* (a non-empty partition reported as empty), the
reducer never
attempts to read that block, so the data checksum path is never exercised.
- SPARK-40872 handles a related but distinct push-based-shuffle case (a
merged chunk that
is empty because it was merged by a bad node) via a fallback to fetching
the original
blocks. It doesn't cover non-empty-to-empty corruption of a mapper's own
`MapStatus`
outside push-based merging.
- SPARK-57491 handles two attempts of the same partition both completing a
push under
speculation/non-deterministic shuffle keys, where only one attempt's
`MapStatus`
should survive. It doesn't address transport/memory corruption of a single
`MapStatus`.
None of the above detects the case where a `MapStatus` that was correct when
the mapper
produced it gets corrupted before a reducer consumes it. This PR closes that
gap.
### Does this PR introduce _any_ user-facing change?
No behavior change by default: both new configs are internal and
`spark.shuffle.mapStatus.checksum.enabled` defaults to `false`.
When explicitly enabled, a `MapStatus` metadata checksum mismatch now
surfaces as a `FetchFailedException`, which triggers Spark's standard
stage-retry recovery, instead of silently under-reading shuffle data.
### How was this patch tested?
- New unit tests in `MapStatusChecksumSuite` cover: `compute`/`recompute`
returning `None` for all-empty inputs, invariance to size values (only
non-empty indices matter), sensitivity to a single-bit change in the non-empty
set, agreement with an independent CRC32 reference implementation, ADLER32 vs
CRC32 producing different values, rejection of unsupported algorithm names,
`recompute` agreeing with `compute` and detecting a changed non-empty set, and
Java serialization round-trips of the new field on both `CompressedMapStatus`
and `HighlyCompressedMapStatus`.
- New test in `MapOutputTrackerSuite`, `"MapStatus checksum mismatch
triggers FetchFailedException"`, exercises `verifyChecksumOrFail` against a
real `MapOutputTrackerMaster` with a corrupted checksum and asserts a
`FetchFailedException` is thrown.
- Ran both suites locally: `build/sbt 'core/testOnly
*MapStatusChecksumSuite'` and `build/sbt 'core/testOnly
*MapOutputTrackerSuite'` -- all tests pass.
### Was this patch authored or co-authored using generative AI tooling?
Yes, Generated-by: Claude Code (claude-sonnet-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]
