From a2f8b53c6478898cadba5c0f106bc55ce8c4eb0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Thu, 23 Jul 2026 15:35:28 +0800 Subject: [PATCH 1/2] fix: avoid KeepAlive NPE when idle fires before WS handshake IdleStateHandler can emit IdleStateEvent after TCP connect but before WebSocket handshake completes. KeepAliveHandler previously dereferenced a null channel field, logging "connection operation failed" and leaving half-dead connections that recovered only after pool retry. Also close the channel in exceptionCaught so the session pool reconnects promptly. Fixes customer intermittent Stream disconnect (Aone 84622857). Co-authored-by: Cursor --- .../app-stream-network-ws/pom.xml | 5 +++ .../stream/network/ws/KeepAliveHandler.java | 31 +++++++++++--- .../stream/network/ws/NettyClientHandler.java | 4 ++ .../network/ws/KeepAliveHandlerTest.java | 42 +++++++++++++++++++ 4 files changed, 76 insertions(+), 6 deletions(-) create mode 100644 app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java diff --git a/app-stream-network/app-stream-network-ws/pom.xml b/app-stream-network/app-stream-network-ws/pom.xml index b045dae..48dd5e5 100644 --- a/app-stream-network/app-stream-network-ws/pom.xml +++ b/app-stream-network/app-stream-network-ws/pom.xml @@ -30,5 +30,10 @@ io.netty netty-handler-proxy + + junit + junit + test + diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java index b2fa344..5f8b81a 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java @@ -31,8 +31,8 @@ public class KeepAliveHandler extends SimpleChannelInboundHandler { private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(KeepAliveHandler.class); private final Duration timeout; - private final static HashedWheelTimer TIMER = new HashedWheelTimer(); - private Channel channel; + private final static HashedWheelTimer TIMER = new HashedWheelTimer(); + private volatile Channel channel; private final Map timeouts; private final AtomicBoolean active; @@ -52,7 +52,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc } if (evt instanceof IdleStateEvent) { - channel.eventLoop().execute(new PingTask()); + // IdleStateHandler starts counting after TCP connect, before WS handshake. + // Never touch the channel field until handshake completes, otherwise NPE: + // connection operation failed (KeepAliveHandler.java:56) + if (active.get()) { + final Channel ch = channel != null ? channel : ctx.channel(); + if (ch != null && ch.isActive()) { + ch.eventLoop().execute(new PingTask(ch)); + } + } } super.userEventTriggered(ctx, evt); } @@ -69,6 +77,8 @@ protected void channelRead0(ChannelHandlerContext ctx, PongWebSocketFrame msg) t @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + active.set(false); + channel = null; shutdown(); super.channelInactive(ctx); } @@ -84,24 +94,33 @@ private void shutdown() { } private class PingTask implements Runnable { + private final Channel target; + + private PingTask(Channel target) { + this.target = target; + } + @Override public void run() { + if (target == null || !target.isActive()) { + return; + } if (!timeouts.isEmpty()) { return; } final String seq = UUID.randomUUID().toString(); ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes()); PingWebSocketFrame frame = new PingWebSocketFrame(byteBuf); - channel.writeAndFlush(frame).addListener(future -> { + target.writeAndFlush(frame).addListener(future -> { if (future.isSuccess()) { Timeout pingTimeout = TIMER.newTimeout(timeout -> { LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); timeouts.remove(seq); - channel.close(); + target.close(); }, timeout.toMillis(), TimeUnit.MILLISECONDS); timeouts.put(seq, pingTimeout); } else { - channel.close(); + target.close(); } }); } diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java index c3a2dbb..71c7812 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java @@ -53,5 +53,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E } else { LOGGER.error("[DingTalk] connection operation failed, connectionId={}", connectionId, cause); } + // Close the channel so session pool can reconnect instead of leaving a half-dead connection. + if (ctx != null && ctx.channel() != null && ctx.channel().isActive()) { + ctx.close(); + } } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java new file mode 100644 index 0000000..cce4ed9 --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -0,0 +1,42 @@ +package com.dingtalk.open.app.stream.network.ws; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import io.netty.handler.timeout.IdleStateEvent; +import org.junit.Assert; +import org.junit.Test; + +import java.time.Duration; + +/** + * Regression for Aone 84622857 / customer logs: + * IdleStateEvent before WS handshake must not NPE in KeepAliveHandler. + */ +public class KeepAliveHandlerTest { + + @Test + public void idleBeforeHandshakeDoesNotThrow() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + try { + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + Assert.assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void idleAfterHandshakeDoesNotThrow() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + Assert.assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } +} From a5f55635bb297c89e19f1376c4a0fd21edca04bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Thu, 23 Jul 2026 15:52:56 +0800 Subject: [PATCH 2/2] test: cover idle-before-handshake and exceptionCaught close Strengthen regression for Aone 84622857 KeepAlive NPE and ensure NettyClientHandler closes channel on connection failures. Co-authored-by: Cursor --- .../network/ws/KeepAliveHandlerTest.java | 3 ++ .../network/ws/NettyClientHandlerTest.java | 40 +++++++++++++++++++ 2 files changed, 43 insertions(+) create mode 100644 app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java index cce4ed9..f444c67 100644 --- a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -19,7 +19,10 @@ public void idleBeforeHandshakeDoesNotThrow() { KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); EmbeddedChannel channel = new EmbeddedChannel(handler); try { + // Multiple idle events before handshake — reproduces Aone 84622857 NPE path. channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT); Assert.assertTrue(channel.isActive()); } finally { channel.finishAndReleaseAll(); diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java new file mode 100644 index 0000000..b93777b --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java @@ -0,0 +1,40 @@ +package com.dingtalk.open.app.stream.network.ws; + +import com.dingtalk.open.app.stream.network.api.ClientConnectionListener; +import com.dingtalk.open.app.stream.network.api.Context; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Ensure exceptionCaught closes the channel so session pool can reconnect. + */ +public class NettyClientHandlerTest { + + @Test + public void exceptionCaughtClosesActiveChannel() { + final AtomicBoolean disconnected = new AtomicBoolean(false); + NettyClientHandler handler = new NettyClientHandler("conn-test", new ClientConnectionListener() { + @Override + public void receive(Context context) { + } + + @Override + public void onDisConnection(String connectionId) { + disconnected.set(true); + } + }); + + EmbeddedChannel channel = new EmbeddedChannel(handler); + Assert.assertTrue(channel.isActive()); + channel.pipeline().fireExceptionCaught(new RuntimeException("simulated io failure")); + + // EmbeddedChannel closes synchronously on close() + Assert.assertFalse("channel should be closed after exceptionCaught", channel.isActive()); + // channelInactive should notify listener + Assert.assertTrue("onDisConnection should be triggered", disconnected.get()); + channel.finishAndReleaseAll(); + } +}