ChuckLin2025 commented on code in PR #58440:
URL: https://github.com/apache/spark/pull/58440#discussion_r3931197988
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java:
##########
@@ -284,6 +290,55 @@ 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.
+ */
Review Comment:
Hi Timothy, here is the background of this PR: in our internal traffic, we
got a report that one query's data leaked to a sibling query which is running
in the cluster at the same time. We confirmed this throught the spark task
event, the number of the sibling query's task fetched shuffle bytes is
identical to a task's written shuffle bytes (this task belongs to another
query). Two stages had the same plan but based on different input sources. Also
the number of bytes here is a 9 digits number so couldn't be a coincident.
The log also printed the to-be-fetched block size which is still correct. So
we suspected this may happens in the transport layer. We reviewed the logic
here carefully. Loically the server side has mutex to guarantee it always send
reply in-order. And the underlying TPC protocal should guarantee the client
side see the data in order as well. We had no exact ideas about why this may
heppens. So in this PR, I claim this is a potential corruption.
We also consider introducing an end-to-end- block checksum or at least
number of bytes validation. But that's a overhead for the Driver memory
pressure. By default spark only stores compressed size of the blockers as the
metadata for poetnial optimizer cases. We have concerns about memory here.
But more imprortant here is this assertion in our internal traffic actually
caught another live desynchronization case (even tough the frequency is quite
low). And it prevented this kind of slient data bug as expected, by closing the
buggy connection and let the task retry the fetch reqeust. So we don't have
plan to add a more solid validataion like checksum. But we believe the
assertion here could also benfit the community
--
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]
