cloud-fan commented on code in PR #57310:
URL: https://github.com/apache/spark/pull/57310#discussion_r3597441563
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -973,6 +976,30 @@ 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 =>
!perReplica.get(bm).contains(sealedValue.get))) {
Review Comment:
When `spark.shuffle.service.fetch.rdd.enabled` is set, `updateBlockInfo`
adds an external-shuffle-service pseudo-`BlockManagerId` to `blockLocations`
for disk RDD blocks (lines 899-902) but never records it in `blockChecksums`.
So for a correctly sealed block, `perReplica.get(essId)` is `None` here and
this guard fires, returning a spurious "tracks a replica whose checksum
differs" diagnostic.
Local-checkpoint DISK_ONLY blocks are disk-persisted RDD blocks, so this is
reachable under a supported config. The seal and read paths already
special-case ESS ids by port (lines 390, 1012), so I'd skip them here too:
```suggestion
} else if (locations.exists(bm =>
bm.port != externalShuffleServicePort &&
!perReplica.get(bm).contains(sealedValue.get))) {
```
Worth a test with ESS RDD fetch enabled, since nothing currently covers that
path.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -3000,7 +3000,7 @@ package object config {
.booleanConf
.createWithDefault(false)
- private[spark] val LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED =
+ val LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED =
Review Comment:
This is the only fully-public `val` in a package object of ~378
`private[spark]` configs, and it's `.internal()`. `private[spark]` already
reaches every `org.apache.spark.*` caller (the current readers in `RDD.scala`
and `LocalCheckpointSuite.scala` are both inside it). Which caller needs full
public visibility — is it OSS, or a downstream fork? If downstream-only, this
widening probably shouldn't land in the OSS PR.
--
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]
