This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b65ef3e18fbe [SPARK-58107][CORE][FOLLOWUP] Keep the local-checkpoint
sealed checksum as a tombstone past last-replica loss
b65ef3e18fbe is described below
commit b65ef3e18fbe2fd499e1c48edd212cae79d377ba
Author: Juliusz Sompolski <[email protected]>
AuthorDate: Fri Jul 24 23:49:44 2026 +0800
[SPARK-58107][CORE][FOLLOWUP] Keep the local-checkpoint sealed checksum as
a tombstone past last-replica loss
### What changes were proposed in this pull request?
Follow-up to
[SPARK-58107](https://issues.apache.org/jira/browse/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)
Closes #57499 from
juliuszsompolski/julek_data/checksum-seal-verify-tombstone.
Authored-by: Juliusz Sompolski <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/storage/BlockManagerMasterEndpoint.scala | 18 +++++--
.../apache/spark/storage/BlockManagerSuite.scala | 57 ++++++++++++++++++++++
2 files changed, 71 insertions(+), 4 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
index 2807ab839bf4..08d802740618 100644
---
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -87,9 +87,15 @@ class BlockManagerMasterEndpoint(
// The authoritative sealed checksum per finalized block. Once present, only
a copy with this
// checksum is admitted to `blockLocations`; divergent copies are rejected,
and reads self-check
- // against it. Kept in lock-step with `blockLocations` on block/rdd/BM
removal.
+ // against it. Unlike `blockChecksums`/`blockLocations`, this is a
tombstone: it survives the loss
+ // of the block's last replica (so a late divergent copy of a finalized
block is still rejected)
+ // and is reclaimed only with the RDD, in `removeRdd`.
private val sealedChecksums = new JHashMap[BlockId, Long]
+ // rddId -> its sealed block ids, letting `removeRdd` reclaim the
`sealedChecksums` tombstones by
+ // rddId without scanning the whole map.
+ private val sealedBlocksByRdd = new mutable.HashMap[Int,
mutable.HashSet[RDDBlockId]]
+
// Mapping from task id to the set of rdd blocks which are generated from
the task.
private val tidToRddBlockIds = new mutable.HashMap[Long,
mutable.HashSet[RDDBlockId]]
// Record the RDD blocks which are not visible yet, a block will be removed
from this collection
@@ -376,7 +382,6 @@ class BlockManagerMasterEndpoint(
blocks.foreach { blockId =>
val bms: mutable.HashSet[BlockManagerId] = blockLocations.remove(blockId)
blockChecksums.remove(blockId)
- sealedChecksums.remove(blockId)
if (trackingCacheVisibility) {
invisibleRDDBlocks.remove(blockId)
}
@@ -401,6 +406,10 @@ class BlockManagerMasterEndpoint(
}
}
}
+
+ // Reclaim this RDD's sealed checksums via the index.
+ sealedBlocksByRdd.remove(rddId).foreach(_.foreach(sealedChecksums.remove))
+
val removeRddFromExecutorsFutures = blockManagerInfo.values.map { bmInfo =>
bmInfo.storageEndpoint.ask[Int](removeMsg).recover {
// use 0 as default value means no blocks were removed
@@ -542,7 +551,7 @@ class BlockManagerMasterEndpoint(
if (locations.isEmpty) {
blockLocations.remove(blockId)
blockChecksums.remove(blockId)
- sealedChecksums.remove(blockId)
+ // `sealedChecksums` deliberately kept (tombstone; see its def).
logWarning(log"No more replicas available for ${MDC(BLOCK_ID,
blockId)}!")
} else if (proactivelyReplicate && (blockId.isRDD ||
blockId.isInstanceOf[TestBlockId])) {
// As a heuristic, assume single executor failure to find out the
number of replicas that
@@ -907,7 +916,7 @@ class BlockManagerMasterEndpoint(
if (locations.isEmpty) {
blockLocations.remove(blockId)
blockChecksums.remove(blockId)
- sealedChecksums.remove(blockId)
+ // `sealedChecksums` deliberately kept (tombstone; see its def).
}
true
}
@@ -959,6 +968,7 @@ class BlockManagerMasterEndpoint(
// because a lost local-checkpoint block cannot be recomputed.
val winner = perReplica.values.groupBy(identity).maxBy(_._2.size)._1
sealedChecksums.put(blockId, winner)
+ sealedBlocksByRdd.getOrElseUpdate(rddId, new
mutable.HashSet[RDDBlockId]) += blockId
val losers = perReplica.collect { case (bmId, c) if c != winner =>
bmId }.toSeq
losers.foreach { bmId =>
perReplica.remove(bmId)
diff --git
a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
index 2241e3ee5a92..e2e2552270f7 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -2580,6 +2580,63 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
assert(master.getSealedChecksum(blockId).isEmpty)
}
+ test("sealed checksum tombstone rejects a divergent late copy after the last
replica is lost") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val blockId = RDDBlockId(20, 0)
+ // store1 materializes and the block is sealed to checksum 42.
+ master.updateBlockInfo(
+ store1.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ assert(master.sealRddChecksums(20) === 0)
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+ // store1's only replica is lost (invalid storage level empties its
locations).
+ master.updateBlockInfo(store1.blockManagerId, blockId, StorageLevel.NONE,
0, 0, Some(42L))
+ assert(master.getLocations(blockId).isEmpty)
+ // The seal is a tombstone: it survives the loss of the last replica.
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+ // A late divergent copy (e.g. a zombie task) is still rejected by the
tombstone.
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(99L))
+ assert(!master.getLocations(blockId).contains(store2.blockManagerId))
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+ // A late copy matching the seal is admitted.
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ assert(master.getLocations(blockId).contains(store2.blockManagerId))
+ }
+
+ test("sealed checksum tombstone survives losing the last replica via
executor removal") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val blockId = RDDBlockId(21, 0)
+ master.updateBlockInfo(
+ store1.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ assert(master.sealRddChecksums(21) === 0)
+ // The only replica's executor dies (removeBlockManager empties the
block's locations).
+ master.removeExecutor(store1.blockManagerId.executorId)
+ assert(master.getLocations(blockId).isEmpty)
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+ // A divergent late copy is still rejected.
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(99L))
+ assert(!master.getLocations(blockId).contains(store2.blockManagerId))
+ }
+
+ test("removeRdd reclaims a sealed checksum tombstone whose last replica was
lost") {
+ val store = makeBlockManager(20000, "exec1")
+ val blockId = RDDBlockId(22, 0)
+ master.updateBlockInfo(
+ store.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100, Some(42L))
+ assert(master.sealRddChecksums(22) === 0)
+ // Lose the last replica; the tombstone remains even though the block has
no location.
+ master.updateBlockInfo(store.blockManagerId, blockId, StorageLevel.NONE,
0, 0, Some(42L))
+ assert(master.getLocations(blockId).isEmpty)
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+ // removeRdd reclaims the tombstone via the sealedBlocksByRdd index (not
blockLocations).
+ master.removeRdd(22, blocking = true)
+ assert(master.getSealedChecksum(blockId).isEmpty)
+ }
+
test("local-checkpoint verification: agreeing disk and _SER replicas survive
the seal") {
val diskStore = makeBlockManager(20000, "exec1")
val serStore = makeBlockManager(20000, "exec2")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
