ZequnLin created SPARK-59142:
--------------------------------
Summary: Verify streamId when matching stream fetch responses to
callbacks in TransportResponseHandler
Key: SPARK-59142
URL: https://issues.apache.org/jira/browse/SPARK-59142
Project: Spark
Issue Type: Bug
Components: Spark Core
Affects Versions: 5.0.0
Reporter: ZequnLin
TransportResponseHandler.handle() matches an incoming StreamResponse /
StreamFailure
to its callback by FIFO streamCallbacks.poll() order. This relies on the
assumption
(from SPARK-11265 / apache/spark#9206) that the server answers StreamRequests
in the
same order the client sent them. The response carries a streamId and the client
stored
the expected streamId when it registered the callback, but the two are never
compared.
If that ordering invariant is ever violated, poll() binds a StreamResponse to
the wrong
callback and silently delivers the wrong block's bytes to the reader. Because
the
StreamResponse wire message carries only (streamId, byteCount) and
StreamInterceptor only
enforces the byte count, a wrong-but-self-consistent block passes every
existing check --
surfacing as shuffle data corruption on the fetch-to-disk path
(OneForOneBlockFetcher ->
client.stream()), or as a task hang for a reader whose response never arrives.
The
chunk-fetch path is unaffected (it keys outstandingFetches by StreamChunkId).
Fix: add a streamId equality check after poll() in both the StreamResponse and
StreamFailure branches. On mismatch, fail the polled callback under its own
streamId (so
its caller does not hang waiting for a response it will never correctly
receive), then
throw IllegalStateException. The throw propagates to Netty's exceptionCaught,
the
connection is torn down, and its remaining outstanding requests are re-fetched
in order on
a fresh channel. This converts a silent, undetectable corruption / hang into a
loud,
retriable fetch failure. The check cannot false-fire: the client registers each
callback
under the exact streamId it requested, so registered streamId == response
streamId is an
invariant of every correct delivery.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
