From b31d7e412023ccaaff339449bcf01f4cd903312a Mon Sep 17 00:00:00 2001 From: datadog-bits <263423550+datadog-bits@users.noreply.github.com> Date: Fri, 30 Oct 2026 08:26:06 +0000 Subject: [PATCH 1/2] Limit Netty write failures to close frames Co-authored-by: cgcote <5115638+cgcote@users.noreply.github.com> --- .../server/BlockingResponseHandler.java | 8 +++-- .../server/BlockingResponseHandlerTest.java | 31 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java index fa73bcd2b99..be9a0190ea4 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java @@ -17,6 +17,7 @@ import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpUtil; import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.util.ReferenceCountUtil; import java.nio.channels.ClosedChannelException; import java.util.Map; @@ -247,8 +248,11 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { log.info("Ignored write of object {}", msg); ReferenceCountUtil.release(msg); - // Releasing the message does not complete the write; notify listeners that it failed. - promise.tryFailure(new ClosedChannelException()); + if (msg instanceof CloseWebSocketFrame) { + // Let Netty's close listener run when a close frame is discarded. Failing other writes + // can re-enter framework error handling after the blocking response (e.g. in Vert.x). + promise.tryFailure(new ClosedChannelException()); + } } } } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java index 5ccba3de956..8d617fb6702 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java @@ -3,6 +3,7 @@ import static datadog.trace.instrumentation.netty41.server.BlockingResponseHandler.BEFORE_BLOCKING_HANDLER_NAME; import static datadog.trace.instrumentation.netty41.server.BlockingResponseHandler.HANDLER_NAME; import static io.netty.handler.codec.http.HttpMethod.GET; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; import static java.util.Collections.emptyMap; import static java.util.concurrent.TimeUnit.SECONDS; @@ -18,6 +19,7 @@ import datadog.context.Context; import datadog.trace.instrumentation.netty41.ServerRequestContext; import datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.NettyBlockResponseFunction; +import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelPromise; @@ -25,16 +27,45 @@ import io.netty.channel.embedded.EmbeddedChannel; import io.netty.channel.local.LocalChannel; import io.netty.handler.codec.http.DefaultFullHttpRequest; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.DefaultLastHttpContent; import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpContent; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.util.concurrent.Future; import java.nio.channels.ClosedChannelException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; class BlockingResponseHandlerTest { + @ParameterizedTest + @MethodSource("discardedHttpWritesDoNotFireExceptionsArguments") + void discardedHttpWritesDoNotFireExceptions(HttpContent message) { + EmbeddedChannel channel = + new EmbeddedChannel(BlockingResponseHandler.IgnoreAllWritesHandler.INSTANCE); + try { + channel.pipeline().writeAndFlush(message, channel.voidPromise()); + + channel.checkException(); + assertEquals(0, message.refCnt()); + assertNull(channel.readOutbound()); + assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + static Stream discardedHttpWritesDoNotFireExceptionsArguments() { + return Stream.of( + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.buffer().writeByte(1)), + new DefaultLastHttpContent(Unpooled.buffer().writeByte(1))); + } + @Test void discardedCloseFrameCompletesPromiseAndNotifiesCloseListener() { EmbeddedChannel channel = From 8c1e5e87a8806c974b3598462eec04df54ca2552 Mon Sep 17 00:00:00 2001 From: datadog-bits <263423550+datadog-bits@users.noreply.github.com> Date: Fri, 30 Oct 2026 13:26:44 +0000 Subject: [PATCH 2/2] Complete discarded Netty write promises Co-authored-by: cgcote <5115638+cgcote@users.noreply.github.com> --- .../server/BlockingResponseHandler.java | 6 +++-- .../server/BlockingResponseHandlerTest.java | 26 +++++++++++++++++-- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java index be9a0190ea4..8772b7d1403 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandler.java @@ -248,10 +248,12 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { log.info("Ignored write of object {}", msg); ReferenceCountUtil.release(msg); + // Fail close frames so Netty's close listener runs. Complete other discarded writes + // successfully to avoid re-entering framework error handling (e.g. in Vert.x). if (msg instanceof CloseWebSocketFrame) { - // Let Netty's close listener run when a close frame is discarded. Failing other writes - // can re-enter framework error handling after the blocking response (e.g. in Vert.x). promise.tryFailure(new ClosedChannelException()); + } else { + promise.trySuccess(); } } } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java index 8d617fb6702..a95264bccc0 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/BlockingResponseHandlerTest.java @@ -20,6 +20,7 @@ import datadog.trace.instrumentation.netty41.ServerRequestContext; import datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.NettyBlockResponseFunction; import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelPromise; @@ -44,7 +45,7 @@ class BlockingResponseHandlerTest { @ParameterizedTest - @MethodSource("discardedHttpWritesDoNotFireExceptionsArguments") + @MethodSource("discardedHttpWritesArguments") void discardedHttpWritesDoNotFireExceptions(HttpContent message) { EmbeddedChannel channel = new EmbeddedChannel(BlockingResponseHandler.IgnoreAllWritesHandler.INSTANCE); @@ -60,7 +61,28 @@ void discardedHttpWritesDoNotFireExceptions(HttpContent message) { } } - static Stream discardedHttpWritesDoNotFireExceptionsArguments() { + @ParameterizedTest + @MethodSource("discardedHttpWritesArguments") + void discardedHttpWritesCompletePromiseAndNotifyListener(HttpContent message) { + EmbeddedChannel channel = + new EmbeddedChannel(BlockingResponseHandler.IgnoreAllWritesHandler.INSTANCE); + AtomicReference notifiedPromise = new AtomicReference<>(); + try { + ChannelFuture future = channel.pipeline().writeAndFlush(message); + future.addListener((ChannelFutureListener) notifiedPromise::set); + + assertTrue(future.isSuccess()); + assertSame(future, notifiedPromise.get()); + channel.checkException(); + assertEquals(0, message.refCnt()); + assertNull(channel.readOutbound()); + assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + static Stream discardedHttpWritesArguments() { return Stream.of( new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.buffer().writeByte(1)), new DefaultLastHttpContent(Unpooled.buffer().writeByte(1)));