yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4268166033
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java:
##########
@@ -396,7 +397,13 @@ private void close(boolean iteratorHasNext) throws
IOException {
writeMetrics.incRecordsWritten(tmpRecordsWritten);
long waitStartTime = System.nanoTime();
- shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers,
numPartitions);
+ shuffleClient.mapperEnd(
+ shuffleId,
+ mapId,
+ encodedAttemptId,
+ numMappers,
+ numPartitions,
+ new MapperMetrics(pusher.getUncompressedBytesWritten()));
Review Comment:
Good catch. `pushGiantRecord` now increments a writer-level giant-bytes
counter that is summed with the pusher's counter in the `MapperMetrics`
reported at mapperEnd, mirroring `HashBasedShuffleWriter`. The writer suites
now assert the reported uncompressed bytes equal the bytes that reached the
shuffle client — the all-giant-record tests (`testGiantRecord`,
`testGiantRecordWithFastWrite`, which force every record above
`pushBufferMaxSize`) fail without the fix, and the mixed-path
`checkWithIntegrity` covers giant + buffered records together.
##########
client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala:
##########
@@ -71,6 +73,17 @@ private[celeborn] class CelebornListener(
mayUpdate()
}
+ override def onOtherEvent(event: SparkListenerEvent): Unit = {
+ event match {
+ case e: CelebornMapperMetricsEvent =>
+ if (pluginEnabled) {
+ totalUncompressedWriteBytes.addAndGet(e.getUncompressedBytesWritten)
Review Comment:
Aligned. Both sides now use the same set of mapper attempts:
- The listener counts `shuffleWriteMetrics` only for successful task ends
(failed/killed attempts push bytes whose MapperEnd is never reported).
- `LifecycleManager` reports metrics for every successfully closed writer
attempt — including speculative/rerun attempts whose MapperEnd the commit
handler rejects — deduplicated by attemptId so a retried MapperEnd RPC of the
same attempt is not counted twice.
Covered by
`CelebornListenerSuite.shuffleWriteMetricsCountOnlySuccessfulAttempts` (success
+ killed attempts + mapper events) and a `LifecycleManagerSuite` test driving
the accepted / retried / duplicate-attempt cases over real RPC.
--
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]
