uros-b commented on code in PR #58440:
URL: https://github.com/apache/spark/pull/58440#discussion_r3904563808
##########
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:
The new method builds its log message with String.format(...) and passes the
result to logger.error(msg) as a plain pre-formatted string. Every other
logger.error/logger.warn call with dynamic arguments in this file (migrated
under SPARK-48209) uses the structured MDC form: logger.error("... {} ...",
MDC.of(LogKeys.STREAM_ID, ...), MDC.of(LogKeys.HOST_PORT, ...)). Both
LogKeys.STREAM_ID and LogKeys.HOST_PORT already exist and are already imported
in this file; the fix is mechanical.
--
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]
