This is an automated email from the ASF dual-hosted git repository.
HuangZhenQiu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4d87a7d2489 [FLINK-40903][runtime/network] Add source host info in TM
disconnect error
4d87a7d2489 is described below
commit 4d87a7d2489886dad1dcdf25843d32d2c83f8da2
Author: Shihuan Liu <[email protected]>
AuthorDate: Mon Oct 12 08:23:41 2026 -0700
[FLINK-40903][runtime/network] Add source host info in TM disconnect error
This closes #29391
---
.../CreditBasedPartitionRequestClientHandler.java | 11 +++-
...editBasedPartitionRequestClientHandlerTest.java | 65 ++++++++++++++++++++++
2 files changed, 73 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
index b5c899dc3ed..1e4161bba67 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -131,6 +131,7 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
final SocketAddress remoteAddr = ctx.channel().remoteAddress();
+ final SocketAddress localAddr = ctx.channel().localAddress();
notifyAllChannelsOfErrorAndClose(
new RemoteTransportException(
@@ -139,7 +140,9 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
+ " [ "
+
connectionID.getResourceID().getStringWithMetadata()
+ " ] "
- + "'. "
+ + "' (local address: '"
+ + localAddr
+ + "'). "
+ "This might indicate that the remote task
manager was lost.",
remoteAddr));
@@ -157,6 +160,7 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
notifyAllChannelsOfErrorAndClose(cause);
} else {
final SocketAddress remoteAddr = ctx.channel().remoteAddress();
+ final SocketAddress localAddr = ctx.channel().localAddress();
final TransportException tex;
@@ -170,12 +174,13 @@ class CreditBasedPartitionRequestClientHandler extends
ChannelInboundHandlerAdap
+ " [ "
+
connectionID.getResourceID().getStringWithMetadata()
+ " ] "
- + "'. "
+ + "' (local address: '"
+ + localAddr
+ + "'). "
+ "This indicates that the remote task
manager was lost.",
remoteAddr,
cause);
} else {
- final SocketAddress localAddr = ctx.channel().localAddress();
tex =
new LocalTransportException(
String.format(
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
index d5801803696..1553ce25afd 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
@@ -54,6 +54,7 @@ import
org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
import org.apache.flink.shaded.netty4.io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import
org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
import
org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
@@ -68,6 +69,7 @@ import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.SocketAddress;
import java.util.stream.Stream;
import static
org.apache.flink.runtime.io.network.netty.PartitionRequestQueueTest.blockChannel;
@@ -728,6 +730,69 @@ class CreditBasedPartitionRequestClientHandlerTest {
cause, expectedClass));
}
+ @Test
+ void testChannelInactiveErrorContainsLocalAndRemoteAddress() {
+ final InetSocketAddress localAddress = new
InetSocketAddress("127.0.0.1", 12345);
+ final InetSocketAddress remoteAddress = new
InetSocketAddress("127.0.0.2", 54321);
+ final CreditBasedPartitionRequestClientHandler handler =
+ new CreditBasedPartitionRequestClientHandler();
+ handler.setConnectionId(new ConnectionID(ResourceID.generate(),
remoteAddress, 0));
+ final EmbeddedChannel channel =
+ createEmbeddedChannelWithAddresses(localAddress,
remoteAddress, handler);
+
+ channel.close();
+
+ assertThatThrownBy(handler::checkError)
+ .isInstanceOf(RemoteTransportException.class)
+ .hasMessageContaining("Connection unexpectedly closed by
remote task manager")
+ .hasMessageContaining(remoteAddress.toString())
+ .hasMessageContaining("local address: '" + localAddress + "'");
+ }
+
+ @Test
+ void testConnectionResetErrorContainsLocalAndRemoteAddress() {
+ final InetSocketAddress localAddress = new
InetSocketAddress("127.0.0.1", 12345);
+ final InetSocketAddress remoteAddress = new
InetSocketAddress("127.0.0.2", 54321);
+ final CreditBasedPartitionRequestClientHandler handler =
+ new CreditBasedPartitionRequestClientHandler();
+ handler.setConnectionId(new ConnectionID(ResourceID.generate(),
remoteAddress, 0));
+ final EmbeddedChannel channel =
+ createEmbeddedChannelWithAddresses(
+ localAddress,
+ remoteAddress,
+ new ChannelInboundHandlerAdapter() {
+ @Override
+ public void channelRead(ChannelHandlerContext ctx,
Object msg)
+ throws Exception {
+ throw new IOException("Connection reset by
peer");
+ }
+ },
+ handler);
+
+ channel.writeInbound(1);
+
+ assertThatThrownBy(handler::checkError)
+ .isInstanceOf(RemoteTransportException.class)
+ .hasMessageContaining("Lost connection to task manager")
+ .hasMessageContaining(remoteAddress.toString())
+ .hasMessageContaining("local address: '" + localAddress + "'");
+ }
+
+ private static EmbeddedChannel createEmbeddedChannelWithAddresses(
+ SocketAddress localAddress, SocketAddress remoteAddress,
ChannelHandler... handlers) {
+ return new EmbeddedChannel(handlers) {
+ @Override
+ protected SocketAddress localAddress0() {
+ return localAddress;
+ }
+
+ @Override
+ protected SocketAddress remoteAddress0() {
+ return remoteAddress;
+ }
+ };
+ }
+
@Test
void testAnnounceBufferSize() throws Exception {
final CreditBasedPartitionRequestClientHandler handler =
