[
https://issues.apache.org/jira/browse/SPARK-59142?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59142:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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]
