ChuckLin2025 opened a new pull request, #58440: URL: https://github.com/apache/spark/pull/58440
### What changes were proposed in this pull request? `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 `StreamRequest`s 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 were never compared. This PR adds a `streamId` equality check after `poll()` in both the `StreamResponse` and `StreamFailure` branches. On a mismatch it: 1. Fails the polled callback (under its own registered `streamId`) with an `IOException`, so its caller does not hang waiting for a response it will never correctly receive (`poll()` has already removed that callback from the queue, so it would otherwise be orphaned). 2. Throws `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. It also emits a dedicated `logger.error` at each detection site whose message contains `desynced` and both `streamId`s and the remote address, so this previously-silent condition is greppable. ### Why are the changes needed? If the FIFO ordering invariant is ever violated, `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)` -- no blockId, no content check -- and `StreamInterceptor` only enforces the server-declared byte count, a wrong-but-self-consistent block passes every existing check. This surfaces as shuffle **data corruption** on the `>maxRemoteBlockSizeFetchToMem` fetch-to-disk path (`OneForOneBlockFetcher` -> `client.stream()` -> the `StreamResponse` branch), or as a **hung task** for a reader whose response never correctly arrives. The chunk-fetch path is unaffected (it keys `outstandingFetches` by `StreamChunkId`). This change converts that silent, undetectable corruption / hang into a loud, retriable fetch failure -- no corrupt bytes, and no hang. 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; a mismatch is always a real desync. That is also why tearing down the connection is correct rather than overkill: once the queue is desynced, every subsequent `poll()` on that channel is suspect, so a fresh in-order connection is required. ### Does this PR introduce _any_ user-facing change? No. On correct executions (the FIFO invariant holds) behavior is unchanged. It only affects the previously-undetected desync case, which it turns from silent wrong results / a hang into an existing, retriable fetch-failure code path; no API or result-schema change. ### How was this patch tested? Added four unit tests to `TransportResponseHandlerSuite`: - `streamResponseWithMismatchedStreamIdThrows` -- a `StreamResponse` whose `streamId` differs from the head-of-queue callback throws `IllegalStateException` (message contains `desynced`), does not deliver success to the wrong callback, and fails the polled callback under its own `streamId`. - `streamFailureWithMismatchedStreamIdThrows` -- same for the `StreamFailure` branch. - `desyncTearsDownConnectionAndFailsAllOutstandingRequestsRetriably` -- on a channel with an innocent concurrent chunk fetch in flight, a desync fails the polled stream callback inline and, via the `exceptionCaught` teardown path, fails the remaining chunk fetch -- all retriably, none receiving data. - `streamResponseWithMatchingStreamIdIsDelivered` -- regression guard: a matching `streamId` is still delivered normally. Ran `build/sbt 'network-common/testOnly org.apache.spark.network.TransportResponseHandlerSuite'`: Passed, 15 total, 0 failed. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 4.8) -- 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]
