Release requests in cors handler (#32364)

There are two scenarios where a http request could terminate in the cors
handler. If that occurs, the requests need to be released. This commit
releases those requests.

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

@@ -24,6 +24,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
@@ -50,7 +51,7 @@ public class Netty4CorsHandler extends ChannelDuplexHandler {
     private static Pattern SCHEME_PATTERN = Pattern.compile("^https?://");
 
     private final Netty4CorsConfig config;
-    private HttpRequest request;
+    private FullHttpRequest request;
 
     /**
      * Creates a new instance with the specified {@link Netty4CorsConfig}.
@@ -64,15 +65,24 @@ public Netty4CorsHandler(final Netty4CorsConfig config) {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (config.isCorsSupportEnabled() && msg instanceof HttpRequest) {
-            request = (HttpRequest) msg;
+        assert msg instanceof FullHttpRequest : "Invalid message type: " + msg.getClass();
+        if (config.isCorsSupportEnabled()) {
+            request = (FullHttpRequest) msg;
             if (isPreflightRequest(request)) {
-                handlePreflight(ctx, request);
-                return;
+                try {
+                    handlePreflight(ctx, request);
+                    return;
+                } finally {
+                    releaseRequest();
+                }
             }
             if (config.isShortCircuit() && !validateOrigin()) {
-                forbidden(ctx, request);
-                return;
+                try {
+                    forbidden(ctx, request);
+                    return;
+                } finally {
+                    releaseRequest();
+                }
             }
         }
         ctx.fireChannelRead(msg);
@@ -123,6 +133,11 @@ private void handlePreflight(final ChannelHandlerContext ctx, final HttpRequest
         }
     }
 
+    private void releaseRequest() {
+        request.release();
+        request = null;
+    }
+
     private static void forbidden(final ChannelHandlerContext ctx, final HttpRequest request) {
         ctx.writeAndFlush(new DefaultFullHttpResponse(request.protocolVersion(), HttpResponseStatus.FORBIDDEN))
             .addListener(ChannelFutureListener.CLOSE);

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/cors/NioCorsHandler.java

@@ -24,6 +24,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
@@ -50,7 +51,7 @@ public class NioCorsHandler extends ChannelDuplexHandler {
     private static Pattern SCHEME_PATTERN = Pattern.compile("^https?://");
 
     private final NioCorsConfig config;
-    private HttpRequest request;
+    private FullHttpRequest request;
 
     /**
      * Creates a new instance with the specified {@link NioCorsConfig}.
@@ -64,15 +65,24 @@ public NioCorsHandler(final NioCorsConfig config) {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (config.isCorsSupportEnabled() && msg instanceof HttpRequest) {
-            request = (HttpRequest) msg;
+        assert msg instanceof FullHttpRequest : "Invalid message type: " + msg.getClass();
+        if (config.isCorsSupportEnabled()) {
+            request = (FullHttpRequest) msg;
             if (isPreflightRequest(request)) {
-                handlePreflight(ctx, request);
-                return;
+                try {
+                    handlePreflight(ctx, request);
+                    return;
+                } finally {
+                    releaseRequest();
+                }
             }
             if (config.isShortCircuit() && !validateOrigin()) {
-                forbidden(ctx, request);
-                return;
+                try {
+                    forbidden(ctx, request);
+                    return;
+                } finally {
+                    releaseRequest();
+                }
             }
         }
         ctx.fireChannelRead(msg);
@@ -109,6 +119,11 @@ public static void setCorsResponseHeaders(HttpRequest request, HttpResponse resp
         }
     }
 
+    private void releaseRequest() {
+        request.release();
+        request = null;
+    }
+
     private void handlePreflight(final ChannelHandlerContext ctx, final HttpRequest request) {
         final HttpResponse response = new DefaultFullHttpResponse(request.protocolVersion(), HttpResponseStatus.OK, true, true);
         if (setOrigin(response)) {