[SPARK-58107][CORE][FOLLOWUP] Keep the local-checkpoint sealed checksum as a tombstone past last-replica loss

[juliuszsompolski]

What changes were proposed in this pull request?

Follow-up to SPARK-58107 (detect and repair
non-deterministic local checkpoints via RDD-block content checksums). Makes the master's per-block
sealed checksum (sealedChecksums in BlockManagerMasterEndpoint) a tombstone that survives the
loss of the block's last replica and is reclaimed only when the RDD is removed.

Previously sealedChecksums(blockId) was dropped alongside blockLocations when a block's last
replica went away (the empty-locations branches of updateBlockInfo and removeBlockManager).

• updateBlockInfo / removeBlockManager: on empty locations, keep sealedChecksums (still drop
  blockLocations / blockChecksums). A late replica matching the seal is admitted, a divergent one
  rejected by checksumSealRejectsUpdate.
• removeRdd: reclaim the RDD's tombstones via a new sealedBlocksByRdd index (rddId -> its sealed
  block ids, populated in sealRddChecksums), so reclaim is O(the RDD's sealed blocks) rather than a
  scan of the whole sealedChecksums map on the single-threaded endpoint.
• Only sealedChecksums (a Long per sealed partition, plus its RDDBlockId in the index) is
  tombstoned; blockChecksums is still dropped at empty-locations.

Why are the changes needed?

After a checkpoint is sealed and its lineage is cut, losing the winning replica erased the seal. A
late divergent copy of that block -- e.g. from a zombie task of a superseded stage attempt that was
still running from before the seal -- then had no seal to reject it, so updateBlockInfo admitted it
and later readers could silently observe a different version than earlier readers. Losing every
replica of a finalized checkpoint must mean data loss, not a silent version switch. The tombstone
preserves the seal so the divergent late copy is still rejected; retention is bounded by RDD lifetime
(reclaimed in removeRdd).

Does this PR introduce any user-facing change?

No. spark.checkpoint.local.verifyChecksum.enabled gates the seal path; behavior only differs when it
is enabled, and only in the (previously silently wrong) last-replica-loss + divergent-late-copy case.

How was this patch tested?

New BlockManagerSuite tests drive the master RPC sequence the race produces (no scheduler needed):
after a sealed block loses its last replica -- via an invalid-level updateBlockInfo and via
removeExecutor -- the seal survives as a tombstone, a divergent late registration is rejected and a
matching one admitted; and removeRdd reclaims a zero-replica tombstone. The existing "sealed
checksums are cleared when the RDD is removed" test still passes.

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

Generated-by: Claude Code (Opus 4.8)

[juliuszsompolski]

cc @cloud-fan

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
The metadata lifetime, rejection path, and RDD-scoped cleanup are internally consistent, with focused regression coverage for the changed behavior.

Verification

Traced sealRddChecksums through both replica-removal paths and removeRdd, confirmed the reverse index is populated only alongside a sealed checksum and drained at RDD removal, and independently checked the new tests against those production paths. The text and contract scanners examined all prepared candidates and reported no findings.

[cloud-fan]

Merge Summary:

• merged into master b65ef3e
• merged into branch-4.x 51c1d14

Posted by merge_spark_pr.py