Release requests in cors handle (#32410) (#32505)

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

@@ -23,6 +23,7 @@
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 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;
@@ -48,7 +49,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}.
@@ -62,15 +63,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);
@@ -113,6 +123,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);