[SPARK-57232][SS][RTM][StreamingShuffle][Part7] Add additional e2e testing for streaming shuffle

[jerrypeng]

What changes were proposed in this pull request?

This is part 7 of the multi-PR effort to add streaming shuffle to Spark -- a push-based shuffle used by Real-Time Mode (RTM) structured streaming, where writer tasks push records directly to reader tasks over the network instead of writing map output to disk for readers to pull. It adds the end-to-end test suite for the writer (part 4) and reader (part 5), together with a small operability improvement to the writer that the suite exercises.

End-to-end tests -- StreamingShuffleSuite, a real-Netty integration suite that drives the full writer <-> reader path now that the writer and reader are merged. It covers, among other things: writer <-> reader data transfer, the credit-control / termination handshake, sequence-number validation (out-of-order / duplicate / missing detection), checksum verification, memory back-pressure, background-thread error propagation via ErrorNotifier, resource cleanup on task completion (including race conditions between background-thread failure and task-thread completion), and that a reader fails (rather than hangs) when a writer disconnects before terminating.

Writer oversized-row warnings -- StreamingShuffleWriter.write() now logs a throttled warning when a serialized row is larger than the network block size, or larger than 25% of the total writer memory. Because a single row is never split across network buffers, an oversized row forces its own buffer past the configured block size and inflates the tracked memory budget; the warning tells operators to raise the block size or writer memory rather than silently overshoot. The throttling reuses the writer's existing LogThrottler, whose signature is adjusted to carry structured log context.

Test-only accessor -- TransportServer.channelFuture() exposes the server's bound channel so the suite can assert the writer's server is up while writing and torn down after task completion.

The full PR stack:

• Part 1 (SPARK-56674, merged) - streaming shuffle wire protocol (the four Netty message types).
• Part 2 (SPARK-56962, merged) - StreamingShuffleOutputTracker (driver-side writer-location coordination).
• Part 3 (SPARK-57141, merged) - shuffle-manager layer (StreamingShuffleManager + MultiShuffleManager).
• Part 3.5 (SPARK-57337, merged) - shared transport + error plumbing (ErrorNotifier, TransportClient.send(ByteBuf), checksum config).
• Part 4 (SPARK-57229, merged) - StreamingShuffleWriter + server-side Netty handler (push path).
• Part 5 (SPARK-57230, merged) - StreamingShuffleReader + client-side Netty handler (pull path).
• Part 7 (this PR) - end-to-end StreamingShuffleSuite + writer oversized-row warnings.
• Remaining parts wire the shuffle into the scheduler/block-manager lifecycle (activation) and add documentation.

With parts 1 through 5 all merged, this PR is standalone against master.

Why are the changes needed?

The writer (part 4) and reader (part 5) were merged with unit-level coverage only. This PR adds the integration coverage that exercises them together over real Netty connections -- the push/pull data path, the credit-control and termination handshakes, error propagation, and resource cleanup -- so regressions in the end-to-end protocol are caught. The oversized-row warnings give operators actionable feedback when row sizes defeat the writer's buffering, which is otherwise silent.

Does this PR introduce any user-facing change?

No. The only production-code change is two additional (throttled) log warnings on the streaming shuffle write path, which is reached only when the opt-in streaming shuffle manager is configured via spark.shuffle.manager. The default sort shuffle is unaffected.

How was this patch tested?

This PR is itself the test suite. StreamingShuffleSuite is an end-to-end integration suite (real Netty writer + reader) covering the full push/pull path, the credit-control / termination handshake, sequence-number and checksum validation, memory back-pressure, error propagation, resource cleanup, and the oversized-row warnings added here. Run with:

build/sbt "core/testOnly org.apache.spark.shuffle.streaming.StreamingShuffleSuite"

Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Code (Claude Opus 4.8)