SteNicholas commented on code in PR #3718:
URL: https://github.com/apache/celeborn/pull/3718#discussion_r3354585857
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -231,31 +245,116 @@ class MapPartitionCommitHandler(
shuffleId,
(k: Int) => ConcurrentHashMap.newKeySet[Integer]())
resultPartitions.add(partitionId)
+
+ if (shuffleIntegrityCheckEnabled) {
+ recordMapPartitionCommitMetadata(
+ shuffleId,
+ partitionId,
+ numPartitions,
+ crc32PerPartition,
+ bytesWrittenPerPartition)
+ }
}
(dataCommitSuccess, false)
}
+ /**
+ * Records a finished mapper's per-subpartition write-side checksums, keyed
by (shuffleId,
+ * mapPartitionId), for later validation in [[finishPartition]]. A
retried/duplicate attempt
+ * overwrites the prior record (last write wins), which is intentional: the
reader validates
+ * against the last committed attempt. Visible for testing.
+ */
+ private[commit] def recordMapPartitionCommitMetadata(
+ shuffleId: Int,
+ mapPartitionId: Int,
+ numPartitions: Int,
+ crc32PerPartition: Array[Int],
+ bytesWrittenPerPartition: Array[Long]): Unit = {
+ if (crc32PerPartition == null || crc32PerPartition.length != numPartitions
||
+ bytesWrittenPerPartition == null || bytesWrittenPerPartition.length !=
numPartitions) {
+ logWarning(
+ s"Skip recording commit metadata for shuffle $shuffleId map partition
$mapPartitionId " +
+ s"because reported checksum arrays do not match numPartitions
$numPartitions.")
+ return
+ }
+ commitMetadataForMapPartition
+ .computeIfAbsent(
+ shuffleId,
+ (_: Int) => JavaUtils.newConcurrentHashMap[Int,
MapPartitionWriteMetadata]())
+ .put(mapPartitionId, MapPartitionWriteMetadata(crc32PerPartition,
bytesWrittenPerPartition))
+ }
+
override def registerShuffle(
shuffleId: Int,
numMappers: Int,
isSegmentGranularityVisible: Boolean,
numPartitions: Int): Unit = {
super.registerShuffle(shuffleId, numMappers, isSegmentGranularityVisible,
numPartitions)
shuffleIsSegmentGranularityVisible.put(shuffleId,
isSegmentGranularityVisible)
+ // The outer metadata map is created lazily by
recordMapPartitionCommitMetadata.
}
override def isSegmentGranularityVisible(shuffleId: Int): Boolean = {
shuffleIsSegmentGranularityVisible.get(shuffleId)
}
+ /**
+ * Validates a map partition read over the consumed `[startSubIndex,
endSubIndex]` range against
+ * the order-independent combination of the write-side checksums over that
range. The params
+ * rename the trait's reducer-oriented signature
(partitionId/startMapIndex/endMapIndex) to map
+ * semantics. A reader reaches stream end only after commit, so the
write-side metadata is already
+ * recorded; missing metadata fails closed.
+ */
Review Comment:
@xumingming, good catch — the trait-level `CommitHandler.finishPartition`
doc still says "Invoked when a reduce partition...", which is now inaccurate
since map partitions reuse it. Updated the trait doc to be
partition-type-agnostic and kept the map-specific detail on this override.
--
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]
