FMX commented on code in PR #3062:
URL: https://github.com/apache/celeborn/pull/3062#discussion_r1944349231
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1683,23 +1687,58 @@ public void onFailure(Throwable e) {
}
}
+ private int sendCommitMetadataBatchForAllPartitions(
+ int shuffleId, int mapId, int attemptId, PushState pushState, int
numMappers, int numPartitions)
+ throws IOException {
+ // This check should not be required, but added as a safeguard since we
already check at callsites, but added as a safeguard.
+ if (!shuffleIntegrityCheckEnabled) {
+ logger.info("shuffleIntegrityCheck disabled. Not sending any commit
metadata.");
+ return 0;
+ }
+
+ Map<Integer, CommitMetadata> metadataMap =
pushState.getCommitMetadataMap();
+
+ if (!shuffleCompressionEnabled) {
+ logger.info("ShuffleCompression disabled. Not sending any metadata
either");
+ return 0;
+ }
+
+ int bytes = 0;
+
+ for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+ CommitMetadata metadata = metadataMap.getOrDefault(partitionId, new
CommitMetadata());
Review Comment:
Here will always get empty commit meta data.
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java:
##########
@@ -377,7 +377,8 @@ private void close() throws IOException,
InterruptedException {
updateRecordsWrittenMetrics();
long waitStartTime = System.nanoTime();
- shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers);
+ int bytesWritten = shuffleClient.mapperEnd(shuffleId, mapId,
encodedAttemptId, numMappers, numMappers);
+ writeMetrics.incBytesWritten(bytesWritten);
Review Comment:
Should not add the bytes written to spark metrics. Because the write metric
has the correct value of written bytes.
##########
common/src/main/java/org/apache/celeborn/common/write/PushState.java:
##########
@@ -32,6 +33,7 @@ public class PushState {
private final int pushBufferMaxSize;
public AtomicReference<IOException> exception = new AtomicReference<>();
private final InFlightRequestTracker inFlightRequestTracker;
+ private final ConcurrentHashMap<Integer, CommitMetadata> commitMetadataMap =
new ConcurrentHashMap<>();
Review Comment:
This map is always empty. Seems that you forget to update this maps.
--
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]
