ChuckLin2025 commented on code in PR #58440:
URL: https://github.com/apache/spark/pull/58440#discussion_r3905090304
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java:
##########
@@ -284,6 +290,48 @@ public void handle(ResponseMessage message) throws
Exception {
}
}
+ /**
+ * Verifies that the callback polled from the head of the FIFO {@link
#streamCallbacks} queue is
+ * the one this stream response/failure is for, by comparing the callback's
registered streamId
+ * with the streamId carried by the response.
+ *
+ * <p>Under correct operation this equality always holds and the method is a
no-op: responses to
+ * {@code StreamRequest}s arrive on a single connection in the order the
client sent them (see
+ * SPARK-11265), and the client registers each callback under the exact
streamId it requested, so
+ * the head of the queue always corresponds to the next response. A mismatch
is therefore not
+ * reachable by any normal client/server interaction; it could only be
produced by memory or
+ * hardware corruption (e.g. a bit flip in the streamId or a corrupted
queue). This is a defensive
+ * check that turns such corruption -- which would otherwise silently
deliver the wrong block's
+ * bytes to a reader -- into a loud, retriable failure.
+ *
+ * <p>On a mismatch it fails the polled callback under its own streamId (so
its caller does not
+ * hang waiting for a response it will never correctly receive; {@code
poll()} has already removed
+ * it from the queue) and throws {@link IllegalStateException}, which
propagates to Netty's
+ * {@code exceptionCaught} so the connection is torn down and its remaining
outstanding requests
+ * are re-fetched in order on a fresh channel.
+ */
+ private void verifyStreamCallbackMatches(
Review Comment:
Good catch, thanks. Fixed in 53ba42f: the desync `logger.error` now uses the
structured MDC form (`MDC.of(LogKeys.STREAM_ID, responseStreamId)`,
`MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel))`), matching the other
calls in this file. Since the `Logger` API takes only `MDC` args for the `{}`
slots and the file uses distinct keys per value (e.g. `HOST_PORT`/`HOST_PORT2`)
rather than repeating one, I kept the structured line to the arriving
`STREAM_ID` + `HOST_PORT` and left the full detail (both streamIds and the
response/failure kind) on the `IOException`/`IllegalStateException`, which is
logged with full context on the `exceptionCaught` teardown path.
--
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]
