[CELEBORN-2479] Show shuffle compression ratio in Spark UI

[yew1eb]

What changes were proposed in this pull request?

Show the shuffle compression ratio (uncompressed/compressed) on the Celeborn tab of the Spark UI.

• common: new extensible PbMapperMetrics message carried by PbMapperEnd as field 10; MapperEnd gains a defaulted MapperMetrics parameter, so future mapper-side metrics go there instead of new top-level fields.
• client: ShuffleClient adds a mapperEnd overload carrying MapperMetrics (default impl ignores it); LifecycleManager forwards the metrics to a registered MapperEndCallback.
• spark: HashBasedShuffleWriter / SortBasedShuffleWriter count the bytes handed to ShuffleClient before compression (including the giant-record direct-push path) and report them at mapperEnd; SparkShuffleManager posts a CelebornMapperMetricsEvent on the driver listener bus, so the value reaches both the live UI and the History Server event log; CelebornListener aggregates the events and CelebornShufflePage renders the ratio, degrading to N/A when not reported (old clients).
• docker: register CelebornPlugin in the docker spark-defaults.conf so the docker-compose cluster exercises the UI end-to-end.

Numerator and denominator use the same set of mapper attempts: the listener counts shuffleWriteMetrics only for successful task ends, and LifecycleManager reports metrics for every successfully closed writer attempt — including speculative/rerun attempts whose MapperEnd the commit handler rejects — deduplicated by attemptId so retried MapperEnd RPCs are not counted twice.

Why are the changes needed?

The Spark UI shows the shuffle bytes written to Celeborn (post-compression), but there is no way to see how much the compression actually saves. Exposing the ratio helps users evaluate codec choices and estimate real network/disk traffic during shuffle.

CELEBORN-2479

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

The Celeborn tab of the Spark UI shows a new summary line: "Compression Ratio (uncompressed/compressed)". Shuffle write bytes/time now exclude failed or killed attempts, matching Spark's own stage accounting.

How was this patch tested?

• UT: UtilsSuite MapperEnd ser/de roundtrip with MapperMetrics; CelebornListenerSuite aggregation, event JSON roundtrip, and failed-attempt exclusion tests; writer suites assert the uncompressed bytes reported at mapperEnd equal the bytes that reached the shuffle client (all-giant-record and mixed cases); LifecycleManagerSuite exercises the accepted / retried / duplicate-attempt callback behavior over real RPC.
• End-to-end on the docker-compose stack (Spark 3.5.9, hash writer, lz4): confirmed the Celeborn tab shows the compression ratio with a real shuffling job.

[zaynt4606]

Thanks for adding the compression-ratio metric. I found two accounting gaps that can make the displayed ratio incorrect. Please address the inline comments and add regression coverage for these paths.

[zaynt4606]

Could we include bytes sent by pushGiantRecord() in this metric? That path pushes directly through shuffleClient.pushDataWithCRC() and increments writeMetrics.bytesWritten, but never updates SortBasedPusher's uncompressed-byte counter, which is the sole source used here. If all records exceed pushBufferMaxSize, this reports zero and the UI shows N/A despite a successful write. Please account for the direct-push path and cover it with a giant-record test.

[yew1eb]

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.

[zaynt4606]

Could the numerator and denominator use the same set of mapper attempts? onTaskEnd adds shuffleWriteMetrics.bytesWritten for every task end with metrics, while LifecycleManager.handleMapperEnd emits this event only when finishMapperAttempt accepts that attempt. If a failed or speculative attempt writes data and a later attempt succeeds, its compressed bytes remain in the denominator but its uncompressed bytes are absent from the numerator, understating the ratio. Please align the accounting and test a retry/duplicate-attempt case.

[yew1eb]

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.