yew1eb commented on code in PR #3872:
URL: https://github.com/apache/celeborn/pull/3872#discussion_r4348055247
##########
client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java:
##########
@@ -288,6 +289,16 @@ private void checkWithIntegrity(
int[] crcPerPartition = pushState.getCRC32PerPartition(true,
numPartitions);
long[] bytesPerPartition = pushState.getBytesWrittenPerPartition(true,
numPartitions);
+ MapperMetrics mapperMetrics = client.getMapperEndMetrics(shuffleId, 0, 0);
+ assertNotNull(mapperMetrics);
+ long totalBytesPushed = 0;
+ for (long bytes : bytesPerPartition) {
+ totalBytesPushed += bytes;
+ }
+ assertEquals(totalBytesPushed, mapperMetrics.uncompressedBytesWritten());
Review Comment:
Good catch — and it was slightly worse than that: the SortBased async path
pushes through `DataPusher`'s bare `pushData` (CRC already recorded at enqueue
time), which bypassed the WithCRC overloads entirely. Fixed in 45ee042a3 by
making the dummy's `pushData`/`mergeData` themselves return `length * 3 / 4` —
like the real client, every push return value is a post-compression byte count
— so all paths agree. The suites now assert `uncompressed == bytes pushed` and
`compressed == Σ 3/4·batch` separately; I verified a temporary swap of the two
`MapperMetrics` fields turns 6 of 13 SortBased tests red.
##########
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:
Agreed, adopted in 90d35c165. Since both ratio sides travel in the same
MapperMetrics payload, reporting only the accepted attempt keeps them aligned,
`finishMapperAttempt` already dedups by mapId (same-attempt retries and
rejected attempts all return false), and the dedup set is gone along with its
expiry cleanup. The regression test now replays your probe (attempts 0, 0, 1, 1
/ 100, 100, 250, 250) and asserts exactly `[100]`.
--
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]
