This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 39889772efec [SPARK-58107][CORE][FOLLOWUP] Add verifyRddChecksumSeal
invariant check for sealed RDD block checksums
39889772efec is described below
commit 39889772efecf15d9141f1f3577670d1a4a3ae0f
Author: Juliusz Sompolski <[email protected]>
AuthorDate: Wed Jul 22 21:01:30 2026 +0800
[SPARK-58107][CORE][FOLLOWUP] Add verifyRddChecksumSeal invariant check for
sealed RDD block checksums
### What changes were proposed in this pull request?
Follow-up to
[SPARK-58107](https://issues.apache.org/jira/browse/SPARK-58107)
(dedup-and-seal for non-deterministic local checkpoints). That change added
`BlockManagerMaster.sealRddChecksums`, which picks one authoritative content
checksum per RDD block, evicts divergent replicas, and rejects future divergent
registrations. This adds a read-only companion to inspect the result of a seal:
- `BlockManagerMaster.verifyRddChecksumSeal(rddId): Option[String]` (with
the endpoint handler and the `VerifyRddChecksumSeal` message). It checks, for
every block of `rddId` in the directory, that the block is sealed and that
every replica the directory tracks for it carries the sealed content checksum.
Returns `None` when the invariant holds, or `Some(diagnostic)` naming the first
block that violates it (present-but-unsealed, or a tracked replica whose
checksum differs from the seal [...]
The check skips external-shuffle-service pseudo-replicas, identified by
`externalShuffleServicePort` (the same way the seal and read paths do). With
`spark.shuffle.service.fetch.rdd.enabled` set, `updateBlockInfo` adds an ESS
pseudo-`BlockManagerId` to `blockLocations` for a disk RDD block but never to
`blockChecksums` — it is a routing alias to the executor's own on-disk file
(same `executorId`/`host`, only the port differs), not an independent copy, so
there is no separate checksum [...]
### Why are the changes needed?
`sealRddChecksums` mutates the block directory (evicts and reseals), but
there was no way to assert afterward that the directory ended up in the
intended consistent state — every tracked replica agreeing on the sealed
checksum. `verifyRddChecksumSeal` gives callers and tests a cheap invariant
check: under correct operation it always returns `None`, so a non-`None` result
is a tripwire for a seal-machinery regression or an unexpected directory state.
The existing seal tests only assert [...]
### Does this PR introduce _any_ user-facing change?
No. `verifyRddChecksumSeal` is an internal master API with no behavior
change.
### How was this patch tested?
`build/sbt "core/testOnly org.apache.spark.storage.BlockManagerSuite"`:
- A new unit test drives the branches directly: no blocks for an RDD ->
`None`; two agreeing replicas after a seal -> `None`; a present-but-unsealed
block -> `Some(...)`.
- The existing seal tests (agreeing disk and `_SER` replicas survive the
seal; divergent copies converge after eviction; a replicated seal-path block)
now additionally assert `verifyRddChecksumSeal === None`, exercising the check
on genuinely sealed block state.
- A new test drives a sealed disk RDD block through an
external-shuffle-service-fetch-enabled master (so `blockLocations` also tracks
the ESS pseudo-replica) and asserts `verifyRddChecksumSeal === None`; it fails
without the port skip.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57310 from juliuszsompolski/oss-local-checkpoint-verify-seal.
Authored-by: Juliusz Sompolski <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit edadb008523340f5f3a0c37a7b634f570837a6ef)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../apache/spark/storage/BlockManagerMaster.scala | 8 +++
.../spark/storage/BlockManagerMasterEndpoint.scala | 30 ++++++++++
.../spark/storage/BlockManagerMessages.scala | 3 +
.../apache/spark/ExternalShuffleServiceSuite.scala | 40 ++++++++++++++
.../apache/spark/storage/BlockManagerSuite.scala | 64 ++++++++++++++++++++++
5 files changed, 145 insertions(+)
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
index 1b27fc334d23..84a70d27047c 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
@@ -147,6 +147,14 @@ class BlockManagerMaster(
driverEndpoint.askSync[Int](SealRddChecksums(rddId))
}
+ /**
+ * Verify every block of a sealed RDD is sealed and the directory tracks
only replicas carrying
+ * the sealed checksum. Returns None if the invariant holds, else a
diagnostic string.
+ */
+ def verifyRddChecksumSeal(rddId: Int): Option[String] = {
+ driverEndpoint.askSync[Option[String]](VerifyRddChecksumSeal(rddId))
+ }
+
/** Get locations of the blockId from the driver */
def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
driverEndpoint.askSync[Seq[BlockManagerId]](GetLocations(blockId))
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 9c59d5ea49fb..2807ab839bf4 100644
---
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -253,6 +253,9 @@ class BlockManagerMasterEndpoint(
case SealRddChecksums(rddId) =>
context.reply(sealRddChecksums(rddId))
+ case VerifyRddChecksumSeal(rddId) =>
+ context.reply(verifyRddChecksumSeal(rddId))
+
case UpdateRDDBlockVisibility(taskId, visible) =>
// This is to report the information that whether rdd blocks computed by
task(with `taskId`)
// can be turned to be visible. This is reported by DAGScheduler right
after task completes.
@@ -971,6 +974,33 @@ class BlockManagerMasterEndpoint(
unchecksummed
}
+ /**
+ * Invariant check for a sealed RDD: every block of `rddId` in the directory
is sealed, and every
+ * replica the directory tracks for it carries the sealed content checksum.
Returns `None` when
+ * the invariant holds, or `Some(diagnostic)` naming the first block that
violates it. Runs on the
+ * message-handler thread, so it observes a consistent snapshot of the
directory and the checksum
+ * maps.
+ */
+ private def verifyRddChecksumSeal(rddId: Int): Option[String] = {
+ val rddBlocks =
blockLocations.asScala.keys.flatMap(_.asRDDId).filter(_.rddId == rddId)
+ rddBlocks.iterator.map { blockId =>
+ val sealedValue = Option(sealedChecksums.get(blockId)).map(_.longValue)
+ val perReplica =
Option(blockChecksums.get(blockId)).getOrElse(mutable.HashMap.empty)
+ val locations =
Option(blockLocations.get(blockId)).getOrElse(mutable.HashSet.empty)
+ if (sealedValue.isEmpty) {
+ Some(s"$blockId is present but not sealed")
+ } else if (locations.exists(bm =>
+ // Skip the external-shuffle-service pseudo-replica: it has no
checksum entry of its own.
+ bm.port != externalShuffleServicePort &&
+ !perReplica.get(bm).contains(sealedValue.get))) {
+ Some(s"$blockId sealed to ${sealedValue.get} but the directory tracks
a replica whose " +
+ s"checksum differs: locations=$locations checksums=$perReplica")
+ } else {
+ None
+ }
+ }.collectFirst { case Some(violation) => violation }
+ }
+
private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
if (blockLocations.containsKey(blockId)) blockLocations.get(blockId).toSeq
else Seq.empty
}
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
index d32895700f43..3c2ee19ffbd7 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
@@ -122,6 +122,9 @@ private[spark] object BlockManagerMessages {
// divergent copies, and reject future divergent registrations.
case class SealRddChecksums(rddId: Int) extends ToBlockManagerMaster
+ // Invariant-check a sealed RDD; see
BlockManagerMasterEndpoint.verifyRddChecksumSeal.
+ case class VerifyRddChecksumSeal(rddId: Int) extends ToBlockManagerMaster
+
case class GetLocations(blockId: BlockId) extends ToBlockManagerMaster
case class GetLocationsAndStatus(blockId: BlockId, requesterHost: String)
diff --git
a/core/src/test/scala/org/apache/spark/ExternalShuffleServiceSuite.scala
b/core/src/test/scala/org/apache/spark/ExternalShuffleServiceSuite.scala
index 1b3fd2ab463a..4688e6a107de 100644
--- a/core/src/test/scala/org/apache/spark/ExternalShuffleServiceSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ExternalShuffleServiceSuite.scala
@@ -238,6 +238,46 @@ class ExternalShuffleServiceSuite extends ShuffleSuite
with Eventually {
}
}
+ test("SPARK-58107: sealed local checkpoint stays verifiable when fetched via
shuffle service") {
+ val confWithRddFetchEnabled = conf.clone
+ .set(config.SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED, true)
+ .set(config.SHUFFLE_SERVICE_FETCH_RDD_ENABLED, true)
+ .set(config.LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED, true)
+ .set(config.EXECUTOR_REMOVE_DELAY.key, "0s")
+ .set(config.DRIVER_BIND_ADDRESS.key, Utils.localHostName())
+ sc = new SparkContext("local-cluster[1,1,1024]", "test",
confWithRddFetchEnabled)
+ sc.env.blockManager.externalShuffleServiceEnabled should equal(true)
+ try {
+ // A serialized-level local checkpoint is fingerprinted and sealed to
one content checksum
+ // per partition on the master.
+ val rdd = sc.parallelize(0 until 100, 2)
+ .map(i => (i, i))
+ .persist(StorageLevel.DISK_ONLY)
+ rdd.localCheckpoint()
+ rdd.count()
+
+ val blockId = RDDBlockId(rdd.id, 0)
+ // With RDD fetch enabled, the master also tracks the block under the
shuffle-service port.
+ eventually(timeout(2.seconds), interval(100.milliseconds)) {
+ val locations = sc.env.blockManager.master.getLocations(blockId)
+ assert(locations.map(_.port).contains(server.getPort))
+ }
+ assert(sc.env.blockManager.master.verifyRddChecksumSeal(rdd.id).isEmpty)
+
+ // After the executor is gone, only the shuffle-service pseudo-replica
remains; the seal
+ // invariant must still hold (the pseudo-replica carries no checksum and
must be skipped).
+ sc.killExecutors(sc.getExecutorIds())
+ eventually(timeout(2.seconds), interval(100.milliseconds)) {
+ val locations = sc.env.blockManager.master.getLocations(blockId)
+ assert(locations.size === 1)
+ assert(locations.map(_.port).contains(server.getPort))
+ }
+ assert(sc.env.blockManager.master.verifyRddChecksumSeal(rdd.id).isEmpty)
+ } finally {
+ rpcHandler.applicationRemoved(sc.conf.getAppId, true)
+ }
+ }
+
test("SPARK-37618: external shuffle service removes shuffle blocks from
deallocated executors") {
for (enabled <- Seq(true, false)) {
// Use local disk reading to get location of shuffle files on disk
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 7e09c17a5f76..20823e4e5b28 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -2512,6 +2512,63 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
assert(master.getSealedChecksum(RDDBlockId(9, 0)).isEmpty)
}
+ test("verifyRddChecksumSeal holds after a normal seal and detects an
unsealed block") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ // No blocks for this rdd: nothing to check.
+ assert(master.verifyRddChecksumSeal(12) === None)
+
+ // Two agreeing replicas, sealed: every tracked replica carries the sealed
checksum -> holds.
+ val sealedBlock = RDDBlockId(12, 0)
+ master.updateBlockInfo(
+ store1.blockManagerId, sealedBlock, StorageLevel.DISK_ONLY, 0, 100,
Some(77L))
+ master.updateBlockInfo(
+ store2.blockManagerId, sealedBlock, StorageLevel.DISK_ONLY, 0, 100,
Some(77L))
+ assert(master.sealRddChecksums(12) === 0)
+ assert(master.verifyRddChecksumSeal(12) === None)
+
+ // A second block present in the directory but never sealed (e.g. sealing
was skipped) is a
+ // violation: it is present but not sealed.
+ val unsealedBlock = RDDBlockId(12, 1)
+ master.updateBlockInfo(
+ store1.blockManagerId, unsealedBlock, StorageLevel.DISK_ONLY, 0, 100,
Some(88L))
+ val violation = master.verifyRddChecksumSeal(12)
+ assert(violation.exists(_.contains("present but not sealed")), violation)
+ // The divergent-tracked-replica branch is defense-in-depth: the directory
rejects a divergent
+ // registration for a sealed block (see the reject test above), so it is
not reachable here.
+ }
+
+ test("verifyRddChecksumSeal ignores the external-shuffle-service
pseudo-replica") {
+ // The ESS pseudo-replica is tracked in blockLocations but has no checksum
entry.
+ conf.set(SHUFFLE_SERVICE_ENABLED, true)
+ conf.set(SHUFFLE_SERVICE_FETCH_RDD_ENABLED, true)
+ val essBlockManagerInfo = new mutable.HashMap[BlockManagerId,
BlockManagerInfo]()
+ val essMaster = new BlockManagerMaster(
+ rpcEnv.setupEndpoint("blockmanager-ess",
+ new BlockManagerMasterEndpoint(rpcEnv, isLocal = true, conf,
liveListenerBus,
+ Some(mock(classOf[ExternalBlockStoreClient])), essBlockManagerInfo,
mapOutputTracker,
+ shuffleManager, isDriver = true)),
+ rpcEnv.setupEndpoint("blockmanagerHeartbeat-ess",
+ new BlockManagerMasterHeartbeatEndpoint(rpcEnv, true,
essBlockManagerInfo)),
+ conf, true)
+ val bmRef = rpcEnv.setupEndpoint("bm-ess", new RpcEndpoint {
+ override val rpcEnv: RpcEnv = BlockManagerSuite.this.rpcEnv
+ override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
+ case _ => context.reply(true)
+ }
+ })
+ val bmId = essMaster.registerBlockManager(
+ BlockManagerId("exec1", "localhost", 1234, None), Array.empty, 2000, 0,
bmRef)
+
+ val blockId = RDDBlockId(13, 0)
+ essMaster.updateBlockInfo(bmId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(55L))
+ // The directory now tracks both the executor and the ESS pseudo-replica
for the block.
+ assert(essMaster.getLocations(blockId).exists(_.port ==
conf.get(SHUFFLE_SERVICE_PORT)))
+ assert(essMaster.sealRddChecksums(13) === 0)
+ // Only the executor replica carries a checksum; the ESS pseudo-replica
must not be flagged.
+ assert(essMaster.verifyRddChecksumSeal(13) === None)
+ }
+
test("sealed checksums are cleared when the RDD is removed") {
val store = makeBlockManager(20000, "exec1")
val blockId = RDDBlockId(10, 0)
@@ -2544,6 +2601,8 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
assert(master.getLocations(blockId).toSet ===
Set(diskStore.blockManagerId, serStore.blockManagerId))
assert(master.getSealedChecksum(blockId).isDefined)
+ // The seal is internally consistent: both tracked replicas carry the
sealed checksum.
+ assert(master.verifyRddChecksumSeal(11) === None)
}
test("UpdateBlockInfo Externalizable round-trip preserves the checksum
field") {
@@ -2590,6 +2649,9 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
// The seal keeps the plurality (dataA, two copies) and evicts the
divergent dataB copy.
assert(master.sealRddChecksums(20) === 0)
assert(master.getLocations(blockId).toSet === Set(store1.blockManagerId,
store3.blockManagerId))
+ // After evicting the minority, the directory is internally consistent:
every tracked replica
+ // carries the sealed checksum.
+ assert(master.verifyRddChecksumSeal(20) === None)
// The diverged executor no longer serves its stale local copy -- it was
evicted, or the
// read-side self-check skips it because its checksum != the sealed one
(so reads converge).
assert(store2.getLocalValues(blockId).isEmpty)
@@ -2698,6 +2760,8 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
assert(master.getLocations(blockId).size === 2)
assert(store1.getLocalValues(blockId).isDefined)
assert(store2.getLocalValues(blockId).isDefined)
+ // Primary and replica are both tracked with the sealed checksum: the
invariant holds.
+ assert(master.verifyRddChecksumSeal(30) === None)
}
test("SPARK-41497: mark rdd block as visible") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
