xumingming commented on code in PR #3718:
URL: https://github.com/apache/celeborn/pull/3718#discussion_r3349277391
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -76,6 +76,19 @@ class MapPartitionCommitHandler(
// shuffleId -> boolean, records whether the shuffle is visible at the
segment level, facilitating future optimization of worker read and write
processes
private val shuffleIsSegmentGranularityVisible =
JavaUtils.newConcurrentHashMap[Int, Boolean]
+ private val shuffleIntegrityCheckEnabled =
conf.clientShuffleIntegrityCheckEnabled
+
+ // Write-side per-subpartition checksums of one finished map partition
(indexed by subpartition).
+ private case class MapPartitionWriteMetadata(crc32: Array[Int],
bytesWritten: Array[Long]) {
+ require(
+ crc32.length == bytesWritten.length,
+ s"crc32 length ${crc32.length} != bytesWritten length
${bytesWritten.length}")
+ }
+
+ // shuffleId -> (mapPartitionId -> write-side metadata).
+ private val commitMetadataForMapPartition =
Review Comment:
This stores CRC and bytes data for every map partition(M), every
sub-partition(N), so the memory usage is O(M * N), I am not familiar with
Flink, not sure whether it would cost too much memory of the JobManager?
##########
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:
The comment better goes to CommitHandler.finishPartition? Currently it still
says: "Invoked when a reduce partition..."
##########
client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/ReadIntegrityTracker.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.celeborn.plugin.flink;
+
+import java.io.IOException;
+import java.util.function.Consumer;
+
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.flink.shaded.netty4.io.netty.buffer.CompositeByteBuf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.CommitMetadata;
+import org.apache.celeborn.plugin.flink.client.FlinkShuffleClientImpl;
+import org.apache.celeborn.plugin.flink.utils.BufferUtils;
+
+/**
+ * Accumulates the read-side {@link CommitMetadata} (CRC32 + byte count) over
a consumed
+ * subpartition range and reports it to the driver at stream end. Each read
path strips its own
+ * framing via {@code accumulate*Buffer} so the hashed bytes match what the
writer hashed (the body
+ * after the batch header).
+ */
+public class ReadIntegrityTracker {
+ private static final Logger LOG =
LoggerFactory.getLogger(ReadIntegrityTracker.class);
+
+ private final FlinkShuffleClientImpl client;
+ private final int shuffleId;
+ private final int partitionId;
+ private final int subPartitionIndexStart;
+ private final int subPartitionIndexEnd;
+ private final CommitMetadata readCommitMetadata = new CommitMetadata();
+ private volatile boolean enabled;
+ // True once report() has run; the report is terminal whether it succeeded
or failed.
+ private volatile boolean reportAttempted = false;
+
+ public ReadIntegrityTracker(
+ FlinkShuffleClientImpl client,
+ int shuffleId,
+ int partitionId,
+ int subPartitionIndexStart,
+ int subPartitionIndexEnd,
+ boolean enabled) {
+ this.client = client;
+ this.shuffleId = shuffleId;
+ this.partitionId = partitionId;
+ this.subPartitionIndexStart = subPartitionIndexStart;
+ this.subPartitionIndexEnd = subPartitionIndexEnd;
+ this.enabled = enabled;
+ }
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ /** Permanently disables tracking for this stream so nothing is reported at
stream end. */
+ void disable() {
+ enabled = false;
+ }
+
+ /** Strips the {@code headerPrefix}-byte batch header; a too-short buffer
disables tracking. */
+ private void accumulatePlainBuffer(ByteBuf flinkBuffer, int headerPrefix) {
+ if (flinkBuffer.readableBytes() < headerPrefix) {
+ disableForUnexpectedFraming(
Review Comment:
There are many disableForUnexpectedFraming calls in this file, why are they
needed? are these scenarios valid in real Flink job?
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -559,7 +564,7 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
StatusCode.SUCCESS.getValue).build()
} else {
response = PbReadReducerPartitionEndResponse.newBuilder().setStatus(
-
+StatusCode.READ_REDUCER_PARTITION_END_FAILED.getValue).setErrorMsg(errorMessage).build()
Review Comment:
Why there was a "+" here before?
--
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]
