[
https://issues.apache.org/jira/browse/SPARK-59142?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ZequnLin updated SPARK-59142:
-----------------------------
Description:
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.
This proposes a defensive check. Under correct operation the streamId equality
always
holds and the check is a no-op: responses arrive on a single connection in the
order
the client sent them, and each callback is registered under the exact streamId
it
requested, so the head of the FIFO queue always corresponds to the next
response. A
mismatch is NOT reachable by any normal client/server interaction -- it could
only be
produced by memory or hardware corruption (for example a bit flip in the
streamId, or
a corrupted queue).
The concern is what happens if such corruption ever does occur. Because the
StreamResponse wire message carries only (streamId, byteCount) and
StreamInterceptor
enforces only the byte count, a wrong-but-self-consistent block would pass every
existing check and be delivered silently to the reader as shuffle data
corruption on
the fetch-to-disk path (OneForOneBlockFetcher -> client.stream()), or leave a
reader's
task hung waiting for a response it never correctly receives. The chunk-fetch
path is
unaffected (it keys outstandingFetches by StreamChunkId).
Fix: add a streamId equality check after poll(), factored into a single helper
used by
both the StreamResponse and StreamFailure branches. On a mismatch, fail the
polled
callback under its own streamId (so its caller does not hang) and 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: since the client registers each
callback
under the exact streamId it requested, registered streamId == response streamId
is an
invariant of every correct delivery, so tearing down the (evidently corrupted)
connection is the safe response.
was:
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.
> 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.
> This proposes a defensive check. Under correct operation the streamId
> equality always
> holds and the check is a no-op: responses arrive on a single connection in
> the order
> the client sent them, and each callback is registered under the exact
> streamId it
> requested, so the head of the FIFO queue always corresponds to the next
> response. A
> mismatch is NOT reachable by any normal client/server interaction -- it could
> only be
> produced by memory or hardware corruption (for example a bit flip in the
> streamId, or
> a corrupted queue).
> The concern is what happens if such corruption ever does occur. Because the
> StreamResponse wire message carries only (streamId, byteCount) and
> StreamInterceptor
> enforces only the byte count, a wrong-but-self-consistent block would pass
> every
> existing check and be delivered silently to the reader as shuffle data
> corruption on
> the fetch-to-disk path (OneForOneBlockFetcher -> client.stream()), or leave a
> reader's
> task hung waiting for a response it never correctly receives. The chunk-fetch
> path is
> unaffected (it keys outstandingFetches by StreamChunkId).
> Fix: add a streamId equality check after poll(), factored into a single
> helper used by
> both the StreamResponse and StreamFailure branches. On a mismatch, fail the
> polled
> callback under its own streamId (so its caller does not hang) and 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: since the client registers each
> callback
> under the exact streamId it requested, registered streamId == response
> streamId is an
> invariant of every correct delivery, so tearing down the (evidently corrupted)
> connection is the safe response.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
