DataDog · ygree · Jul 13, 2026 · Jul 14, 2026 · Jul 14, 2026 · Jul 14, 2026
@@ -8,6 +8,7 @@
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.noopContinuation;
 import static datadog.trace.instrumentation.netty41.AttributeKeys.CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY;
 import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY;
+import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_STREAM_CODEC_ATTRIBUTE_KEY;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
 import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
@@ -237,6 +238,7 @@ else if (handler instanceof HttpClientCodec) {
               pipeline, handler, HttpClientResponseTracingHandler.INSTANCE);
         } else if (NettyHttp2Helper.isHttp2FrameCodec(handler)) {
           if (NettyHttp2Helper.isServer(handler)) {
+            pipeline.channel().attr(HTTP2_STREAM_CODEC_ATTRIBUTE_KEY).set(Boolean.TRUE);
             NettyPipelineHelper.addHandlerAfter(
                 pipeline,
                 handler,

@@ -15,10 +15,13 @@
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpRequest;
+import java.util.Deque;
 
 @ChannelHandler.Sharable
 public class HttpServerRequestTracingHandler extends ChannelInboundHandlerAdapter {
   public static HttpServerRequestTracingHandler INSTANCE = new HttpServerRequestTracingHandler();
+  private static final String INCOMPLETE_RESPONSE_MESSAGE =
+      "Channel closed before response completed";
 
   @Override
   public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
@@ -54,7 +57,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
       DECORATE.onRequest(span, channel, request, parentContext);
 
       final ServerRequestContext serverContext =
-          ServerRequestContext.add(channel, context, headers.get("accept"));
+          ServerRequestContext.add(channel, context, request);
 
       Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction();
       if (rba != null) {
@@ -89,9 +92,47 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
       super.channelInactive(ctx);
     } finally {
       try {
-        ServerRequestContext.closeAll(ctx.channel());
+        final Deque<ServerRequestContext> storedContexts =
+            ServerRequestContext.removeAll(ctx.channel());
+        if (storedContexts != null) {
+          ServerRequestContext storedContext;
+          while ((storedContext = storedContexts.pollFirst()) != null) {
+            if (storedContext.isResponseStarted()) {
+              finishSpanOnChannelClose(storedContext);
+            } else {
+              publishSpanOnChannelClose(storedContext.tracingContext());
+            }
+          }
+        }
       } catch (final Throwable ignored) {
       }
     }
   }
+
+  private static void finishSpanOnChannelClose(final ServerRequestContext serverContext) {
+    final Context storedContext = serverContext.tracingContext();
+    final AgentSpan span = AgentSpan.fromContext(storedContext);
+    if (span == null) {
+      return;
+    }
+    if (!serverContext.isResponseCloseDelimited()) {
+      // The response declared a body via Content-Length or chunked encoding, but the channel closed
+      // before that body completed. Close-delimited responses, in contrast, end normally when the
+      // connection closes.
+      DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE));
+    }
+    if (!serverContext.isBeforeFinishCalled()) {
+      serverContext.markBeforeFinishCalled();
+      DECORATE.beforeFinish(storedContext);
+    }
+    span.finish();
+  }
+
+  private static void publishSpanOnChannelClose(final Context storedContext) {
+    final AgentSpan span = AgentSpan.fromContext(storedContext);
+    if (span != null && span.phasedFinish()) {
+      // At this point we can just publish this span to avoid losing the rest of the trace.
+      span.publish();
+    }
+  }
 }
@@ -1,5 +1,6 @@
 package datadog.trace.instrumentation.netty41.server;
 
+import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_STREAM_CODEC_ATTRIBUTE_KEY;
 import static datadog.trace.instrumentation.netty41.AttributeKeys.WEBSOCKET_SENDER_HANDLER_CONTEXT;
 import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE;
 
@@ -8,25 +9,27 @@
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext;
 import datadog.trace.instrumentation.netty41.ServerRequestContext;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufHolder;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.FileRegion;
 import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.concurrent.Future;
 
 @ChannelHandler.Sharable
 public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter {
   public static HttpServerResponseTracingHandler INSTANCE = new HttpServerResponseTracingHandler();
 
   @Override
   public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) {
-    if (!(msg instanceof HttpResponse)) {
-      ctx.write(msg, prm);
-      return;
-    }
-
     final ServerRequestContext serverContext = ServerRequestContext.nextResponse(ctx.channel());
     final Context storedContext = serverContext == null ? null : serverContext.tracingContext();
     final AgentSpan span = AgentSpan.fromContext(storedContext);
@@ -36,33 +39,165 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann
       return;
     }
 
-    try (final ContextScope scope = storedContext.attach()) {
-      final HttpResponse response = (HttpResponse) msg;
+    if (!(msg instanceof HttpResponse) && !serverContext.isResponseStarted()) {
+      ctx.write(msg, prm);
+      return;
+    }
 
+    try (final ContextScope ignored = storedContext.attach()) {
+      final boolean terminalResponse = isTerminalResponse(ctx, span, serverContext, msg);
+      final boolean knownLengthResponseComplete =
+          !terminalResponse && serverContext.recordResponseBodyBytes(responseBodyBytes(msg));
+      final boolean finishResponseOnWrite = terminalResponse || knownLengthResponseComplete;
+      final ChannelPromise writePromise =
+          finishResponseOnWrite && prm.isVoid() ? ctx.newPromise() : prm;
       try {
-        ctx.write(msg, prm);
+        if (finishResponseOnWrite) {
+          ServerRequestContext.remove(ctx.channel(), serverContext);
+          writePromise.addListener(
+              future -> finishSpan(serverContext, storedContext, span, future));
+        }
+        ctx.write(msg, writePromise);
       } catch (final Throwable throwable) {
-        DECORATE.onError(span, throwable);
-        span.setHttpStatusCode(500);
-        span.finish(); // Finish the span manually since finishSpanOnClose was false
-        ServerRequestContext.remove(ctx.channel(), serverContext);
+        if (!finishResponseOnWrite || !writePromise.isDone()) {
+          DECORATE.onError(span, throwable);
+          span.setHttpStatusCode(500);
+          if (!finishResponseOnWrite) {
+            ServerRequestContext.remove(ctx.channel(), serverContext);
+          }
+          finishSpan(serverContext, storedContext, span);
+        }
         throw throwable;
       }
-      final boolean isWebsocketUpgrade =
-          response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS
-              && "websocket".equals(response.headers().get(HttpHeaderNames.UPGRADE));
+    }
+  }
+
+  private static boolean isTerminalResponse(
+      final ChannelHandlerContext ctx,
+      final AgentSpan span,
+      final ServerRequestContext serverContext,
+      final Object msg) {
+    if (msg instanceof HttpResponse) {
+      final HttpResponse response = (HttpResponse) msg;
+
+      final boolean isWebsocketUpgrade = isWebsocketUpgrade(response);
+      if (isInformationalResponse(response) && !isWebsocketUpgrade) {
+        return false;
+      }
+
       if (isWebsocketUpgrade) {
         ctx.channel()
             .attr(WEBSOCKET_SENDER_HANDLER_CONTEXT)
             .set(new HandlerContext.Sender(span, ctx.channel().id().asShortText()));
       }
-      if (response.status() != HttpResponseStatus.CONTINUE
-          && (response.status() != HttpResponseStatus.SWITCHING_PROTOCOLS || isWebsocketUpgrade)) {
-        DECORATE.onResponse(span, response);
-        DECORATE.beforeFinish(scope.context());
-        span.finish(); // Finish the span manually since finishSpanOnClose was false
-        ServerRequestContext.remove(ctx.channel(), serverContext);
+      DECORATE.onResponse(span, response);
+      serverContext.markResponseStarted();
+      if (isBodylessResponse(serverContext, response) || isWebsocketUpgrade) {
+        return true;
+      }
+      final long contentLength = contentLength(response);
+      if (contentLength > 0 && !HttpUtil.isTransferEncodingChunked(response)) {
+        serverContext.setResponseContentLength(contentLength);
+      }
+      // HTTP/1.x responses with neither a Content-Length nor chunked transfer-encoding are
+      // delimited by the connection closing. HTTP/2 streams translated through
+      // Http2StreamFrameToHttpObjectCodec are delimited by END_STREAM/LastHttpContent instead.
+      if (isCloseDelimitedHttp1Response(ctx, response)) {
+        serverContext.markResponseCloseDelimited();
+        return false;
       }
+      if (msg instanceof LastHttpContent) {
+        return true;
+      }
+      return false;
+    }
+    return serverContext.isResponseStarted()
+        && !serverContext.isResponseCloseDelimited()
+        && msg instanceof LastHttpContent;
+  }
+
+  private static boolean isInformationalResponse(final HttpResponse response) {
+    final int statusCode = response.status().code();
+    return statusCode >= 100 && statusCode < 200;
+  }
+
+  private static boolean isWebsocketUpgrade(final HttpResponse response) {
+    return response.status().code() == HttpResponseStatus.SWITCHING_PROTOCOLS.code()
+        && response
+            .headers()
+            .containsValue(HttpHeaderNames.UPGRADE, HttpHeaderValues.WEBSOCKET, true);
+  }
+
+  private static boolean isBodylessResponse(
+      final ServerRequestContext serverContext, final HttpResponse response) {
+    final int statusCode = response.status().code();
+    return serverContext.isHeadRequest()
+        || statusCode == 204
+        || statusCode == 205
+        || statusCode == 304
+        || (serverContext.isConnectRequest() && statusCode >= 200 && statusCode < 300)
+        || (contentLength(response) == 0 && !HttpUtil.isTransferEncodingChunked(response));
+  }
+
+  private static boolean hasKnownBodyLength(final HttpResponse response) {
+    return contentLength(response) >= 0 || HttpUtil.isTransferEncodingChunked(response);
+  }
+
+  private static boolean isCloseDelimitedHttp1Response(
+      final ChannelHandlerContext ctx, final HttpResponse response) {
+    return !hasKnownBodyLength(response)
+        && !Boolean.TRUE.equals(ctx.channel().attr(HTTP2_STREAM_CODEC_ATTRIBUTE_KEY).get());
+  }
+
+  private static long responseBodyBytes(final Object msg) {
+    if (msg instanceof ByteBuf) {
+      return ((ByteBuf) msg).readableBytes();
+    }
+    if (msg instanceof ByteBufHolder) {
+      return ((ByteBufHolder) msg).content().readableBytes();
+    }
+    if (msg instanceof FileRegion) {
+      return ((FileRegion) msg).count();
+    }
+    return 0;
+  }
+
+  /**
+   * Returns the response {@code Content-Length}, or {@code -1} when it is absent or malformed. A
+   * malformed value is left for Netty's encoder to reject rather than failing the write from the
+   * tracing handler.
+   */
+  private static long contentLength(final HttpResponse response) {
+    try {
+      return HttpUtil.getContentLength(response, -1L);
+    } catch (final NumberFormatException e) {
+      return -1L;
+    }
+  }
+
+  private static void finishSpan(
+      final ServerRequestContext serverContext,
+      final Context storedContext,
+      final AgentSpan span,
+      final Future<?> future) {
+    if (!future.isSuccess()) {
+      DECORATE.onError(span, future.cause());
+      span.setHttpStatusCode(500);
+    }
+    finishSpan(serverContext, storedContext, span);
+  }
+
+  private static void finishSpan(
+      final ServerRequestContext serverContext, final Context storedContext, final AgentSpan span) {
+    beforeFinish(serverContext, storedContext);
+    span.finish(); // Finish the span manually since finishSpanOnClose was false
+  }
+
+  private static void beforeFinish(
+      final ServerRequestContext serverContext, final Context storedContext) {
+    if (!serverContext.isBeforeFinishCalled()) {
+      serverContext.markBeforeFinishCalled();
+      DECORATE.beforeFinish(storedContext);
     }
   }
 }
@@ -63,7 +63,14 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr
       return;
     }
     HttpResponse origResponse = (HttpResponse) msg;
-    if (origResponse.status().code() == HttpResponseStatus.CONTINUE.code()) {
+    int statusCode = origResponse.status().code();
+    // Interim 1xx responses (e.g. 100 Continue, 103 Early Hints) precede the final response.
+    // Analyzing one here would consume the one-shot response analysis before the final response is
+    // written, so its status and headers would never be inspected. Switching Protocols (101) is
+    // terminal, so it is still analyzed.
+    if (statusCode >= 100
+        && statusCode < 200
+        && statusCode != HttpResponseStatus.SWITCHING_PROTOCOLS.code()) {
       super.write(ctx, msg, prm);
       return;
     }

@@ -19,13 +19,13 @@ void disablesTrackingWhenPipeliningLimitIsExceeded() {
     DefaultAttributeMap attributes = new DefaultAttributeMap();
 
     for (int i = 0; i < PIPELINING_LIMIT; i++) {
-      assertNotNull(ServerRequestContext.add(attributes, Context.root(), null));
+      assertNotNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders()));
     }
 
     assertFalse(ServerRequestContext.canTrackRequest(attributes));
     assertNull(ServerRequestContext.nextResponse(attributes));
     assertNull(attributes.attr(AttributeKeys.CONTEXT_ATTRIBUTE_KEY).get());
-    assertNull(ServerRequestContext.add(attributes, Context.root(), null));
+    assertNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders()));
   }
 
   @Test
@@ -35,7 +35,7 @@ void capturesAcceptHeaderValue() {
     headers.set("accept", "text/html");
 
     ServerRequestContext serverContext =
-        ServerRequestContext.add(attributes, Context.root(), headers.get("accept"));
+        ServerRequestContext.add(attributes, Context.root(), headers);
     headers.set("accept", "application/json");
 
     assertEquals("text/html", serverContext.acceptHeader());
@@ -44,7 +44,8 @@ void capturesAcceptHeaderValue() {
   @Test
   void tracksBlockedResponseUntilChannelClose() {
     DefaultAttributeMap attributes = new DefaultAttributeMap();
-    ServerRequestContext serverContext = ServerRequestContext.add(attributes, Context.root(), null);
+    ServerRequestContext serverContext =
+        ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders());
 
     ServerRequestContext.markResponseBlocked(attributes);
     ServerRequestContext.remove(attributes, serverContext);

@@ -9,14 +9,16 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
 import org.junit.jupiter.api.Test;
 
 class MaybeBlockResponseHandlerTest {
 
   @Test
   void dropsWritesAfterBlockedContextHasBeenRemoved() {
     EmbeddedChannel channel = new EmbeddedChannel(MaybeBlockResponseHandler.INSTANCE);
-    ServerRequestContext serverContext = ServerRequestContext.add(channel, Context.root(), null);
+    ServerRequestContext serverContext =
+        ServerRequestContext.add(channel, Context.root(), new DefaultHttpHeaders());
     ServerRequestContext.markResponseBlocked(channel);
     ServerRequestContext.remove(channel, serverContext);
     ByteBuf lateResponseChunk = Unpooled.buffer().writeByte(1);