ChuckLin2025 commented on code in PR #58440:
URL: https://github.com/apache/spark/pull/58440#discussion_r4118225618
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java:
##########
@@ -284,6 +290,52 @@ public void handle(ResponseMessage message) throws
Exception {
}
}
+ /**
+ * Asserts the FIFO ordering invariant for stream responses: the streamId
carried by this
+ * response/failure must equal the streamId of the callback at the head of
the
+ * {@link #streamCallbacks} queue. Responses to {@code StreamRequest}s are
matched to callbacks by
+ * {@code poll()} order (see SPARK-11265), which is correct only if the
server answers requests in
+ * the order the client sent them; this verifies that invariant before the
polled callback is
+ * used. The check cannot false-fire: the client registers each callback
under the exact streamId
+ * it requested, so a matching response always finds its callback at the
head of the queue.
+ *
+ * <p>If the invariant does not hold the callback queue is desynced --
delivering this response
+ * would route the wrong block's bytes to the callback. It fails the polled
callback under its
+ * own streamId with an {@link IOException} (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 block
+ * fetches retry an {@code IOException} via {@code RetryingBlockTransferor})
and throws
+ * {@link IllegalStateException}. The throw propagates to Netty's {@code
exceptionCaught}, which
+ * closes the desynced connection and fails its remaining outstanding
requests; those callers
+ * then apply their own retry behavior where they support it. This turns a
silent wrong-block
+ * delivery into a loud failure.
+ */
+ private void verifyStreamCallbackMatches(
+ Pair<String, StreamCallback> entry, String responseStreamId, String
kind) {
+ if (entry.getLeft().equals(responseStreamId)) {
+ return;
+ }
+ // Log at the detection site, in the structured MDC form used throughout
this file, so this
+ // otherwise-silent condition is directly greppable ("desynced") for
incidence analysis,
+ // independent of the generic connection-exception log the thrown
IllegalStateException produces
+ // downstream.
+ logger.error("Stream callback queue desynced: received streamId {} does
not match the head of "
+ + "the callback queue from {}; failing the connection to avoid
delivering the wrong block.",
+ MDC.of(LogKeys.STREAM_ID, responseStreamId),
+ MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
+ // Full detail (both streamIds) goes on the exception that fails the
callback and tears down the
+ // connection.
+ String msg = String.format(
+ "Stream callback queue desynced: %s streamId %s does not match the head
of the callback "
+ + "queue (streamId %s) from %s; failing the connection to avoid
delivering the wrong "
+ + "block.", kind, responseStreamId, entry.getLeft(),
getRemoteAddress(channel));
+ try {
+ entry.getRight().onFailure(entry.getLeft(), new IOException(msg));
+ } catch (IOException ioe) {
+ logger.warn("Error in stream failure handler.", ioe);
+ }
+ throw new IllegalStateException(msg);
Review Comment:
Fixed in 660fe96. I now build a single `IOException desync = new
IOException(msg)`, fail the polled callback with it, and `throw new
IllegalStateException(msg, desync)` — so the throwable `exceptionCaught`
forwards to the connection's other outstanding callbacks carries an
`IOException` cause, and `RetryingBlockTransferor` retries a concurrent block
fetch instead of failing it permanently. The teardown test now asserts the
collateral fetch's failure carries an `IOException` cause.
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java:
##########
@@ -284,6 +290,52 @@ public void handle(ResponseMessage message) throws
Exception {
}
}
+ /**
+ * Asserts the FIFO ordering invariant for stream responses: the streamId
carried by this
+ * response/failure must equal the streamId of the callback at the head of
the
+ * {@link #streamCallbacks} queue. Responses to {@code StreamRequest}s are
matched to callbacks by
+ * {@code poll()} order (see SPARK-11265), which is correct only if the
server answers requests in
+ * the order the client sent them; this verifies that invariant before the
polled callback is
+ * used. The check cannot false-fire: the client registers each callback
under the exact streamId
+ * it requested, so a matching response always finds its callback at the
head of the queue.
+ *
+ * <p>If the invariant does not hold the callback queue is desynced --
delivering this response
+ * would route the wrong block's bytes to the callback. It fails the polled
callback under its
+ * own streamId with an {@link IOException} (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 block
+ * fetches retry an {@code IOException} via {@code RetryingBlockTransferor})
and throws
+ * {@link IllegalStateException}. The throw propagates to Netty's {@code
exceptionCaught}, which
+ * closes the desynced connection and fails its remaining outstanding
requests; those callers
+ * then apply their own retry behavior where they support it. This turns a
silent wrong-block
+ * delivery into a loud failure.
+ */
+ private void verifyStreamCallbackMatches(
+ Pair<String, StreamCallback> entry, String responseStreamId, String
kind) {
+ if (entry.getLeft().equals(responseStreamId)) {
+ return;
+ }
+ // Log at the detection site, in the structured MDC form used throughout
this file, so this
+ // otherwise-silent condition is directly greppable ("desynced") for
incidence analysis,
+ // independent of the generic connection-exception log the thrown
IllegalStateException produces
+ // downstream.
+ logger.error("Stream callback queue desynced: received streamId {} does
not match the head of "
+ + "the callback queue from {}; failing the connection to avoid
delivering the wrong block.",
+ MDC.of(LogKeys.STREAM_ID, responseStreamId),
+ MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
+ // Full detail (both streamIds) goes on the exception that fails the
callback and tears down the
+ // connection.
+ String msg = String.format(
+ "Stream callback queue desynced: %s streamId %s does not match the head
of the callback "
+ + "queue (streamId %s) from %s; failing the connection to avoid
delivering the wrong "
+ + "block.", kind, responseStreamId, entry.getLeft(),
getRemoteAddress(channel));
+ try {
+ entry.getRight().onFailure(entry.getLeft(), new IOException(msg));
+ } catch (IOException ioe) {
Review Comment:
Fixed in 660fe96. `DownloadCallback.onFailure` now runs
`failRemainingBlocks(...)` (and the temp-file delete) in a `finally`, so the
block is always failed even if `channel.close()` throws — the callback has
already been polled off the response handler's queue, so teardown can't
re-notify it, and skipping the notification would strand the block. The cleanup
error still propagates. Added
`OneForOneBlockFetcherSuite.testFetchToDiskFailsBlockWhenChannelCloseThrowsOnFailure`
covering a channel whose `close()` throws.
##########
common/network-common/src/test/java/org/apache/spark/network/TransportResponseHandlerSuite.java:
##########
@@ -196,6 +198,114 @@ public void
failStreamCallbackWhenInstallingInterceptorFails() throws Exception
assertEquals(0, handler.numOutstandingRequests());
}
+ @Test
+ public void streamResponseWithMismatchedStreamIdThrows() throws Exception {
+ // The FIFO streamCallbacks queue matches responses to callbacks by poll()
order, which is
+ // correct only if a response's streamId equals the head-of-queue
callback's registered streamId
+ // (the server answers StreamRequests in send order; SPARK-11265). This
asserts that invariant:
+ // a StreamResponse whose streamId does not match means the queue is
desynced, and delivering it
+ // would feed the wrong block's bytes to the callback. The check must
throw instead
+ // (SPARK-59142); throwing propagates to Netty's exceptionCaught, which
closes the desynced
+ // connection and fails its remaining outstanding requests.
+ // A mismatch is simulated by registering "stream-A" and handling a
response for "stream-B".
+ Channel c = new LocalChannel();
+ c.pipeline().addLast(TransportFrameDecoder.HANDLER_NAME, new
TransportFrameDecoder());
+ TransportResponseHandler handler = new TransportResponseHandler(c);
+
+ StreamCallback cb = mock(StreamCallback.class);
+ handler.addStreamCallback("stream-A", cb);
+
+ // A response for a different streamId arrives at the head of the FIFO
queue.
+ StreamResponse mismatched = new StreamResponse("stream-B", 1234L, null);
+ IllegalStateException e = assertThrows(IllegalStateException.class,
+ () -> handler.handle(mismatched));
+ assertTrue(e.getMessage().contains("desynced"),
+ "expected a desync error, got: " + e.getMessage());
+ // The mismatched response must NOT have been delivered to the wrong
callback as success...
+ verify(cb, never()).onComplete(any());
+ // ...and the polled callback is failed (with its OWN streamId) so its
caller does not hang.
+ verify(cb, times(1)).onFailure(eq("stream-A"), isA(IOException.class));
+ }
+
+ @Test
+ public void streamFailureWithMismatchedStreamIdThrows() throws Exception {
+ Channel c = new LocalChannel();
+ c.pipeline().addLast(TransportFrameDecoder.HANDLER_NAME, new
TransportFrameDecoder());
+ TransportResponseHandler handler = new TransportResponseHandler(c);
+
+ StreamCallback cb = mock(StreamCallback.class);
+ handler.addStreamCallback("stream-A", cb);
+
+ StreamFailure mismatched = new StreamFailure("stream-B", "uh-oh");
+ IllegalStateException e = assertThrows(IllegalStateException.class,
+ () -> handler.handle(mismatched));
+ assertTrue(e.getMessage().contains("desynced"),
+ "expected a desync error, got: " + e.getMessage());
+ // The failure must NOT be routed under the wrong (response) streamId; the
polled callback is
+ // failed under its OWN streamId instead, so its caller does not hang.
+ verify(cb, never()).onFailure(eq("stream-B"), any());
+ verify(cb, times(1)).onFailure(eq("stream-A"), isA(IOException.class));
+ }
+
+ @Test
+ public void desyncTearsDownConnectionAndFailsAllOutstandingRequests() throws
Exception {
+ // Upstream impact of the streamId assert. In production, throwing from
handle() propagates to
+ // TransportChannelHandler.exceptionCaught, which calls
responseHandler.exceptionCaught (failing
+ // EVERY outstanding request on the channel) and then ctx.close(). This
test simulates that
+ // sequence and shows the meaning for callers: when a stream-callback
desync is detected, the
+ // whole (poisoned) connection is torn down and ALL its in-flight requests
-- the mismatched
+ // stream AND any innocent concurrent chunk-fetch sharing the channel --
are failed. None
+ // receive data. The polled stream callback is failed with an IOException
(which block fetches
+ // retry via RetryingBlockTransferor); the remaining callbacks are failed
on teardown and their
+ // callers apply their own retry behavior where they support it.
+ Channel c = new LocalChannel();
+ c.pipeline().addLast(TransportFrameDecoder.HANDLER_NAME, new
TransportFrameDecoder());
+ TransportResponseHandler handler = new TransportResponseHandler(c);
+
+ // An innocent chunk fetch is in flight on the same connection.
+ StreamChunkId chunkId = new StreamChunkId(1, 0);
+ ChunkReceivedCallback chunkCb = mock(ChunkReceivedCallback.class);
+ handler.addFetchRequest(chunkId, chunkCb);
+ // ...and a stream fetch for "stream-A".
+ StreamCallback streamCb = mock(StreamCallback.class);
+ handler.addStreamCallback("stream-A", streamCb);
+ assertEquals(2, handler.numOutstandingRequests());
+
+ // A StreamResponse for the wrong streamId arrives -> handle() throws
(desync detected).
+ // The desynced (polled) stream callback is failed inline so its caller
does not hang.
+ IllegalStateException thrown = assertThrows(IllegalStateException.class,
+ () -> handler.handle(new StreamResponse("stream-B", 1234L, null)));
+ assertTrue(thrown.getMessage().contains("desynced"));
+ verify(streamCb, times(1)).onFailure(eq("stream-A"), any());
+ verify(streamCb, never()).onComplete(any());
+
+ // Netty then invokes exceptionCaught with the thrown cause; this is the
teardown path that
+ // fails the connection's REMAINING outstanding requests (the innocent
concurrent chunk fetch).
+ handler.exceptionCaught(thrown);
Review Comment:
Fixed in 660fe96. Rewrote
`desyncTearsDownConnectionAndFailsAllOutstandingRequests` to drive the
mismatched `StreamResponse` through a real `TransportChannelHandler` on an
`EmbeddedChannel` (rather than calling `responseHandler.exceptionCaught` by
hand). It now asserts both that the connection's remaining outstanding request
is failed — with an `IOException`-caused (retryable) throwable — and that the
channel is closed (`!channel.isOpen()`).
--
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]
