jerrypeng commented on PR #56878: URL: https://github.com/apache/spark/pull/56878#issuecomment-4898517216
In regards to the nits: > No writer-side timeout on a reader that connects but never acks (or never connects): both the newBuffer() back-pressure loop and the final allAcksReceived.await loop spin indefinitely, escaping only via ErrorNotifier or external task cancellation. If that's intended (relying on task/stage timeouts), a one-line note would help. This is intentional and load-bearing for correctness: the term-ack is the writer's only positive confirmation that a reader received through the final sequence number (onTerminationAckReceived validates the reader's last-seen seq against what we sent). Finishing write() on a timeout without all acks would let the map task report success while a reader is silently missing data — so there's deliberately no timeout. The loop exits on all-acks, an ErrorNotifier error, or task cancellation; and a reader that dies fails its own reduce task, which restarts the query and tears this writer down with it. I've expanded the comment to state this rationale. > startShuffleServer() ignores the Boolean return of registerShuffleWriterTask — a false (shuffle unregistered concurrently) goes unnoticed; worth at least a warning log. The false return isn't silently swallowed: the tracker's registerShuffleWriterTask already logWarnings when the shuffle isn't registered. A false only occurs when the shuffle is being (concurrently) unregistered, and in that case this writer task is being torn down anyway, so there's nothing actionable to do here. I've added a comment at the call site documenting why the return is intentionally not checked. > StreamingShuffleServerHandler's constructor param is named numReaders but the caller passes numPartitions; consistent naming would read better. numReaders is deliberate here — the handler's futureClients array is indexed by shuffleReaderId, so "readers" is the accurate noun for this side; the writer passes its numPartitions because there's exactly one reader per reduce partition. I've added a comment tying the two together rather than renaming, since renaming to numPartitions would be less accurate for the handler's role. > A few stray trailing semicolons in StreamingShuffleServerHandler (Scala doesn't need them). will fix > Two buffer-allocation strategies coexist — data buffers use Unpooled.directBuffer + a custom bufferPool free-list, while the message-envelope composite uses the server's pooled allocator. Reasonable trade-off, but a one-line comment on why they differ would aid future maintainers. Added a comment at the bufferPool declaration explaining the split: fixed-size data payloads use the recyclable direct-buffer free-list, while small variable-size message envelopes use the server's pooled allocator. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
