ForVic commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4343519444
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -980,6 +983,13 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
numPartitions = numPartitions,
crc32PerPartition = crc32PerPartition,
bytesWrittenPerPartition = bytesWrittenPerPartition)
+ // Report every finished writer attempt, including ones the commit handler
rejects
+ // (duplicate speculative attempts); a rejected MapperEnd with the same
attemptId is
+ // just a retried RPC.
+ if (mapperAttemptFinishedSuccess ||
+ attemptId != commitManager.getMapperAttempts(shuffleId)(mapId)) {
+ mapperEndCallback.foreach(_.onMapperEnd(shuffleId, mapId, attemptId,
metrics))
Review Comment:
Now that compressed bytes also come through MapperEnd, do we still need to
report rejected attempts?
If we only report when mapperAttemptFinishedSuccess is true, both sides of
the ratio still use the same attempt set, and finishMapperAttempt already
dedups by mapId. Then this set can go away.
It's a boxed Long per mapper on the driver, and it's populated even when no
callback is registered.
--
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]
