From e08592dc9a4f7ca84c65cf4bfef70ac131f6c753 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 28 Oct 2026 14:39:15 -0700 Subject: [PATCH 1/2] Reproduce StacklessClosedChannelException during a delayed terminal response write --- ...NettyClosedChannelClientAbortSpanTest.java | 134 ++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyClosedChannelClientAbortSpanTest.java diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyClosedChannelClientAbortSpanTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyClosedChannelClientAbortSpanTest.java new file mode 100644 index 00000000000..59706a23e3f --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyClosedChannelClientAbortSpanTest.java @@ -0,0 +1,134 @@ +package datadog.trace.instrumentation.netty41.server; + +import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING; +import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.core.DDSpan; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.DefaultHttpContent; +import io.netty.handler.codec.http.DefaultHttpResponse; +import io.netty.handler.codec.http.HttpRequest; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.LastHttpContent; +import java.net.Socket; +import java.nio.channels.ClosedChannelException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class NettyClosedChannelClientAbortSpanTest extends NettyHttpServerTestSupport { + + private final BlockingQueue delayedWrites = new LinkedBlockingQueue<>(); + + @Override + protected void configurePipeline(Channel channel) { + DelayedWriteHandler delayedWriteHandler = new DelayedWriteHandler(); + channel.pipeline().addLast(delayedWriteHandler); + channel.pipeline().addLast(new HttpServerCodec()); + channel + .pipeline() + .addLast( + new SimpleChannelInboundHandler() { + @Override + protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(TRANSFER_ENCODING, CHUNKED); + ctx.write(response); + ctx.writeAndFlush(new DefaultHttpContent(ctx.alloc().buffer(1).writeByte('a'))); + + // Hold the encoded terminal write after the tracing handler attaches its listener. + delayedWriteHandler.delayWrites = true; + ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT); + } + }); + } + + @Test + void closedChannelFromCancelledResponseDoesNotMarkServerSpanError() throws Exception { + DelayedWrite delayedWrite = null; + try { + try (Socket socket = connect()) { + socket + .getOutputStream() + .write("GET /closed-channel HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(US_ASCII)); + socket.getOutputStream().flush(); + assertTrue(readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 200 ")); + + delayedWrite = delayedWrites.poll(5, SECONDS); + assertNotNull(delayedWrite, "server did not attempt the terminal response write"); + assertFalse(delayedWrite.promise.isDone()); + assertTrue(writer.isEmpty(), "span finished before the terminal response write"); + } + + assertTrue( + delayedWrite.context.channel().closeFuture().await(5, SECONDS), + "server did not observe the client disconnect"); + delayedWrite.resume(); + assertTrue(delayedWrite.promise.await(5, SECONDS), "terminal response write did not finish"); + // Recent Netty versions produce StacklessClosedChannelException; older ones use its base + // type. + Throwable failure = delayedWrite.promise.cause(); + assertInstanceOf(ClosedChannelException.class, failure); + + writer.waitForTraces(1); + assertEquals(1, writer.firstTrace().size()); + DDSpan span = writer.firstTrace().get(0); + assertEquals("netty.request", span.getOperationName().toString()); + assertEquals(200, span.getTag(Tags.HTTP_STATUS)); + assertEquals( + 0, + span.getError(), + () -> "client abort must not count as a trace error: " + failure.getClass().getName()); + } finally { + if (delayedWrite != null && !delayedWrite.promise.isDone()) { + delayedWrite.resume(); + } + } + } + + private final class DelayedWriteHandler extends ChannelOutboundHandlerAdapter { + private boolean delayWrites; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + if (delayWrites) { + delayedWrites.add(new DelayedWrite(ctx, msg, promise)); + } else { + ctx.write(msg, promise); + } + } + } + + private static final class DelayedWrite { + private final ChannelHandlerContext context; + private final Object message; + private final ChannelPromise promise; + + private DelayedWrite(ChannelHandlerContext context, Object message, ChannelPromise promise) { + this.context = context; + this.message = message; + this.promise = promise; + } + + private void resume() throws Exception { + // Let Netty fail the actual write to the closed channel and release the encoded buffer. + context.executor().submit(() -> context.writeAndFlush(message, promise)).sync(); + } + } +} From 3cb116161a9bffdebb93b1b42c55be4b1313b97a Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 28 Oct 2026 15:40:13 -0700 Subject: [PATCH 2/2] Fix Netty client disconnects being reported as trace errors --- .../HttpServerResponseTracingHandler.java | 15 ++++- .../server/NettyChunkedResponseSpanTest.java | 63 +++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java index 342490dfc62..f70bc7f554d 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java @@ -19,6 +19,7 @@ import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.LastHttpContent; import io.netty.util.concurrent.Future; +import java.nio.channels.ClosedChannelException; @ChannelHandler.Sharable public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter { @@ -124,12 +125,24 @@ private static void finishSpan( final Context storedContext, final AgentSpan span, final Future future) { - if (!future.isSuccess()) { + // A client disconnect can fail the terminal write after the response has started. Leave any + // existing span error intact, and retain failures with an underlying cause or another subtype. + if (!future.isSuccess() && !isClosedChannelWithoutCause(future.cause())) { DECORATE.onError(span, future.cause()); } finishSpan(serverContext, storedContext, span); } + private static boolean isClosedChannelWithoutCause(final Throwable failure) { + // StacklessClosedChannelException is package-private and absent from older Netty versions, + // so match it by class name. + return failure != null + && failure.getCause() == null + && (failure.getClass() == ClosedChannelException.class + || "io.netty.channel.StacklessClosedChannelException" + .equals(failure.getClass().getName())); + } + private static void finishSpan( final ServerRequestContext serverContext, final Context storedContext, final AgentSpan span) { try (final ContextScope ignored = storedContext.attach()) { diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java index 1fae3626f0e..cce01ab6eab 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -2,6 +2,7 @@ import static datadog.trace.agent.test.assertions.SpanMatcher.span; import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; import static io.netty.handler.codec.http.HttpHeaderNames.CONNECTION; import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH; import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING; @@ -24,6 +25,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.api.DDTags; import datadog.trace.api.function.TriConsumer; import datadog.trace.api.gateway.Events; import datadog.trace.api.gateway.Flow; @@ -55,6 +57,8 @@ import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.net.Socket; +import java.nio.channels.ClosedByInterruptException; +import java.nio.channels.ClosedChannelException; import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; @@ -68,6 +72,8 @@ import java.util.regex.Pattern; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; +import org.tabletest.junit.TableTest; +import org.tabletest.junit.TypeConverter; @TestInstance(TestInstance.Lifecycle.PER_CLASS) public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { @@ -385,6 +391,63 @@ public Flow apply(RequestContext context, IGSpanInfo span) { } } + @TableTest({ + "Scenario | Write failure | Status | Application error | Expected error", + "plain closed channel | closed | 200 | false | false ", + "existing application error | closed | 200 | true | true ", + "HTTP error response | closed | 500 | false | true ", + "closed channel with underlying failure | closed with cause | 200 | false | true ", + "interrupted operation | interrupted | 200 | false | true ", + "wrapped closed channel | wrapped closed | 200 | false | true " + }) + void preservesImportantErrorsOnTerminalWriteFailure( + Throwable writeFailure, int status, boolean applicationError, boolean expectedError) { + HoldingOutboundHandler holdingHandler = new HoldingOutboundHandler(); + EmbeddedChannel channel = + new EmbeddedChannel(holdingHandler, HttpServerResponseTracingHandler.INSTANCE); + AgentSpan span = startSpan("netty", "netty.request"); + ServerRequestContext.add(channel, span, null); + if (applicationError) { + span.addThrowable(new IllegalStateException("application failure")); + } + + try { + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(status)); + response.headers().set(CONTENT_LENGTH, 0); + channel.writeOutbound(response); + + holdingHandler.failWrite(writeFailure); + assertThrows(writeFailure.getClass(), channel::checkException); + + assertEquals(status, span.getTag(Tags.HTTP_STATUS)); + assertTraces(trace(span().root().operationName("netty.request").error(expectedError))); + if (applicationError) { + assertEquals(IllegalStateException.class.getName(), span.getTag(DDTags.ERROR_TYPE)); + assertEquals("application failure", span.getTag(DDTags.ERROR_MSG)); + } + } finally { + holdingHandler.failWrite(writeFailure); + channel.finishAndReleaseAll(); + } + } + + @TypeConverter + public static Throwable toWriteFailure(String value) { + switch (value) { + case "closed": + return new ClosedChannelException(); + case "closed with cause": + return new ClosedChannelException().initCause(new IOException("transport failure")); + case "interrupted": + return new ClosedByInterruptException(); + case "wrapped closed": + return new IOException("response failure", new ClosedChannelException()); + default: + throw new IllegalArgumentException("Unknown write failure: " + value); + } + } + @Test void knownLengthFileRegionResponseWaitsForLastContent() throws Exception { boolean reportedBeforeLastContent;