[
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, so a simple queue rather than a streamId ->
callback
map suffices. The response carries a streamId and the client stored the expected
streamId when it registered the callback, but the two are never compared.
This adds that comparison as an explicit assertion of the FIFO ordering
invariant. If
the response's streamId does not equal the head-of-queue callback's registered
streamId, the callback queue is desynced.
Why it matters: today such a violation is undetected. poll() binds a
StreamResponse to
the wrong callback and silently delivers the wrong block's bytes to a reader.
Because
the StreamResponse wire message carries only (streamId, byteCount) and
StreamInterceptor
enforces only the byte count, a wrong-but-self-consistent block passes every
existing
check -- surfacing as shuffle data delivered to the wrong reader on the
fetch-to-disk
path (OneForOneBlockFetcher -> client.stream()), or as a hung task whose
response never
correctly arrives. 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 violation, fail the
polled
callback under its own streamId (so its caller does not hang) and throw
IllegalStateException, which propagates to Netty's exceptionCaught so the
connection is
torn down and its remaining outstanding requests are re-fetched in order on a
fresh
channel. This converts a silent wrong-block delivery / hang into a loud,
retriable
fetch failure. The check cannot false-fire: the client registers each callback
under
the exact streamId it requested.
Scope: this asserts the stream-callback ordering invariant; it is deliberately
not a
general data-corruption detector. A payload corruption that leaves the streamId
intact
(e.g. a bit flip in the block bytes) is out of scope -- that would require an
end-to-end block checksum or an authenticated transport (TLS), which this
change does
not add.
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.
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.
> 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, so a simple queue rather than a streamId ->
> callback
> map suffices. The response carries a streamId and the client stored the
> expected
> streamId when it registered the callback, but the two are never compared.
> This adds that comparison as an explicit assertion of the FIFO ordering
> invariant. If
> the response's streamId does not equal the head-of-queue callback's registered
> streamId, the callback queue is desynced.
> Why it matters: today such a violation is undetected. poll() binds a
> StreamResponse to
> the wrong callback and silently delivers the wrong block's bytes to a reader.
> Because
> the StreamResponse wire message carries only (streamId, byteCount) and
> StreamInterceptor
> enforces only the byte count, a wrong-but-self-consistent block passes every
> existing
> check -- surfacing as shuffle data delivered to the wrong reader on the
> fetch-to-disk
> path (OneForOneBlockFetcher -> client.stream()), or as a hung task whose
> response never
> correctly arrives. 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 violation, fail the
> polled
> callback under its own streamId (so its caller does not hang) and throw
> IllegalStateException, which propagates to Netty's exceptionCaught so the
> connection is
> torn down and its remaining outstanding requests are re-fetched in order on a
> fresh
> channel. This converts a silent wrong-block delivery / hang into a loud,
> retriable
> fetch failure. The check cannot false-fire: the client registers each
> callback under
> the exact streamId it requested.
> Scope: this asserts the stream-callback ordering invariant; it is
> deliberately not a
> general data-corruption detector. A payload corruption that leaves the
> streamId intact
> (e.g. a bit flip in the block bytes) is out of scope -- that would require an
> end-to-end block checksum or an authenticated transport (TLS), which this
> change does
> not add.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
