This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7c59784e1d1d [SPARK-57709][CORE] Fail the stream callback and close
the channel when installing the stream interceptor fails
7c59784e1d1d is described below
commit 7c59784e1d1dec43fb8d140d3eb7bf454cc6e1e8
Author: YangJie <[email protected]>
AuthorDate: Mon Jun 29 10:30:33 2026 +0800
[SPARK-57709][CORE] Fail the stream callback and close the channel when
installing the stream interceptor fails
### What changes were proposed in this pull request?
In `TransportResponseHandler`, when a `StreamResponse` arrives and
installing the stream interceptor fails (the `TransportFrameDecoder` is missing
from the channel pipeline, or `setInterceptor` throws), the handler now fails
the stream callback and closes the channel. Previously it only logged the error
and called `deactivateStream()`, leaving the `StreamCallback`, which had
already been removed from the queue, without any completion or failure
notification.
### Why are the changes needed?
This is defensive hardening rather than a fix for a failure seen in
practice. The catch block is effectively unreachable in normal operation: while
an interceptor is active the frame decoder stops framing, so the
`setInterceptor` state check cannot trip, and the decoder is not removed
mid-dispatch on the single channel event loop. Even so, every other branch in
`handle()` notifies its callback while this one did not, so if the path were
ever reached the caller would block until the co [...]
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added
`TransportResponseHandlerSuite.failStreamCallbackWhenInstallingInterceptorFails`,
which handles a `StreamResponse` on a channel whose pipeline has no
`TransportFrameDecoder`, so installing the interceptor throws. It verifies that
the callback is failed with the install exception and that the channel is
closed. The test fails on the old code (the callback receives no interaction)
and passes with the fix; the full `TransportResponseHandlerSuite` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56802 from LuciferYang/streamresponse-callback-leak.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../network/client/TransportResponseHandler.java | 8 +++++++
.../network/TransportResponseHandlerSuite.java | 25 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git
a/common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java
b/common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java
index d27fa08d829b..870be0b561ca 100644
---
a/common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java
+++
b/common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java
@@ -247,6 +247,14 @@ public class TransportResponseHandler extends
MessageHandler<ResponseMessage> {
} catch (Exception e) {
logger.error("Error installing stream handler.", e);
deactivateStream();
+ try {
+ callback.onFailure(resp.streamId, e);
+ } catch (IOException ioe) {
+ logger.warn("Error in stream failure handler.", ioe);
+ }
+ // Installing the interceptor failed, so incoming data on this
channel can no longer
+ // be decoded. Close it so the broken connection is not reused
from the pool.
+ channel.close();
}
} else {
try {
diff --git
a/common/network-common/src/test/java/org/apache/spark/network/TransportResponseHandlerSuite.java
b/common/network-common/src/test/java/org/apache/spark/network/TransportResponseHandlerSuite.java
index ca4bc0a90dfb..7726e9f9b965 100644
---
a/common/network-common/src/test/java/org/apache/spark/network/TransportResponseHandlerSuite.java
+++
b/common/network-common/src/test/java/org/apache/spark/network/TransportResponseHandlerSuite.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import io.netty.channel.Channel;
+import io.netty.channel.ChannelPipeline;
import io.netty.channel.local.LocalChannel;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -171,6 +172,30 @@ public class TransportResponseHandlerSuite {
verify(cb).onFailure(eq("stream-1"), isA(IOException.class));
}
+ @Test
+ public void failStreamCallbackWhenInstallingInterceptorFails() throws
Exception {
+ // With no TransportFrameDecoder in the pipeline, the decoder lookup in
the StreamResponse
+ // branch returns null and installing the interceptor throws. The handler
must fail the
+ // callback (so the caller does not hang) and close the channel rather
than reuse a
+ // connection it can no longer decode.
+ Channel c = mock(Channel.class);
+ ChannelPipeline pipeline = mock(ChannelPipeline.class);
+ when(c.pipeline()).thenReturn(pipeline);
+ when(pipeline.get(TransportFrameDecoder.HANDLER_NAME)).thenReturn(null);
+ TransportResponseHandler handler = new TransportResponseHandler(c);
+
+ StreamCallback cb = mock(StreamCallback.class);
+ handler.addStreamCallback("stream", cb);
+ assertEquals(1, handler.numOutstandingRequests());
+
+ // byteCount > 0 so the handler takes the interceptor-installation path.
+ handler.handle(new StreamResponse("stream", 1234L, null));
+
+ verify(cb, times(1)).onFailure(eq("stream"),
isA(NullPointerException.class));
+ verify(c, times(1)).close();
+ assertEquals(0, handler.numOutstandingRequests());
+ }
+
@Test
public void handleSuccessfulMergedBlockMeta() throws Exception {
TransportResponseHandler handler = new TransportResponseHandler(new
LocalChannel());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
