[7.x] ThreadContext response header leak

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java

@@ -13,6 +13,7 @@
 import io.netty.channel.SimpleChannelInboundHandler;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.http.HttpPipelinedRequest;
+import org.elasticsearch.transport.Transports;
 
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelinedRequest> {
@@ -25,6 +26,8 @@ class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelined
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest httpRequest) {
+        assert Transports.assertDefaultThreadContext(serverTransport.getThreadPool().getThreadContext());
+        assert Transports.assertTransportThread();
         final Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         boolean success = false;
         try {
@@ -40,6 +43,8 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest http
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         ExceptionsHelper.maybeDieOnAnotherThread(cause);
+        assert Transports.assertDefaultThreadContext(serverTransport.getThreadPool().getThreadContext());
+
         Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         if (cause instanceof Error) {
             serverTransport.onException(channel, new Exception(cause));

server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java

@@ -533,4 +533,8 @@ private static ActionListener<Void> earlyResponseListener(HttpRequest request, H
             return NO_OP;
         }
     }
+
+    public ThreadPool getThreadPool() {
+        return threadPool;
+    }
 }

server/src/main/java/org/elasticsearch/http/DefaultRestChannel.java

@@ -125,6 +125,9 @@ public void sendResponse(RestResponse restResponse) {
 
             ActionListener<Void> listener = ActionListener.wrap(() -> Releasables.close(toClose));
             httpChannel.sendResponse(httpResponse, listener);
+//            try (ThreadContext.StoredContext existing = threadContext.stashContext()) {
+//                httpChannel.sendResponse(httpResponse, listener);
+//            }
             success = true;
         } finally {
             if (success == false) {

server/src/main/java/org/elasticsearch/transport/Transports.java

@@ -53,7 +53,11 @@ public static boolean assertNotTransportThread(String reason) {
     public static boolean assertDefaultThreadContext(ThreadContext threadContext) {
         assert threadContext.getRequestHeadersOnly().isEmpty() ||
             threadContext.getRequestHeadersOnly().size() == 1 && threadContext.getRequestHeadersOnly().containsKey(Task.X_OPAQUE_ID) :
-            "expected empty context but was " + threadContext.getRequestHeadersOnly() + " on " + Thread.currentThread().getName();
+            "[request headers] expected empty context but was " + threadContext.getRequestHeadersOnly()
+                + " "+threadContext.getResponseHeaders() + " on " + Thread.currentThread().getName();
+        assert threadContext.getResponseHeaders().isEmpty() :
+            "[response headers] expected empty context but was " + threadContext.getResponseHeaders()
+                + " on " + Thread.currentThread().getName();
         return true;
     }
 }