restore threadcontext for streaming

Signed-off-by: Hailong Cui <ihailong@amazon.com>

Author: Hailong-am

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4HttpServerTransport.java

@@ -17,6 +17,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.BigArrays;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
+import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.common.util.net.NetUtils;
 import org.opensearch.core.common.unit.ByteSizeUnit;
@@ -392,7 +393,10 @@ protected Publisher<Void> incomingRequest(HttpServerRequest request, HttpServerR
                 .subscribe(consumer, error -> {}, () -> consumer.accept(DefaultLastHttpContent.EMPTY_LAST_CONTENT));
 
             incomingStream(new ReactorNetty4HttpRequest(request), consumer.httpChannel());
-            return response.sendObject(consumer);
+
+            final ThreadContext.StoredContext storedContext = threadPool.getThreadContext().newStoredContext(false);
+            // restore ThreadContext when streaming is finished
+            return response.sendObject(consumer).then(Mono.fromRunnable(storedContext::close));
         } else {
             final ReactorNetty4NonStreamingRequestConsumer<HttpContent> consumer = new ReactorNetty4NonStreamingRequestConsumer<>(
                 this,