Handle inbound exceptions on streaming request APIs (elastic#113303)

Today if we submit a request with e.g. invalid credentials to the
`/_bulk` API then we throw a `ClassCastException` trying to cast the
`DefaultHttpRequest` to a `FullHttpRequest` on the error path. This
commit fixes the problem.

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

@@ -46,7 +46,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         assert msg instanceof HttpObject;
         if (msg instanceof HttpRequest request) {
             var preReq = HttpHeadersAuthenticatorUtils.asHttpPreRequest(request);
-            aggregating = decider.test(preReq) && IGNORE_TEST.test(preReq);
+            aggregating = (decider.test(preReq) && IGNORE_TEST.test(preReq)) || request.decoderResult().isFailure();
         }
         if (aggregating || msg instanceof FullHttpRequest) {
             super.channelRead(ctx, msg);

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/security/10_forbidden.yml

@@ -0,0 +1,31 @@
+---
+"Test basic response with invalid credentials":
+
+  - skip:
+      features: headers
+
+  - do:
+      headers:
+        Authorization: "Basic dGVzdF91c2VyOndyb25nLXBhc3N3b3Jk" # invalid credentials
+      info: {}
+      catch: unauthorized
+
+  - match:
+      error.root_cause.0.type: security_exception
+
+---
+"Test bulk response with invalid credentials":
+
+  - skip:
+      features: headers
+
+  - do:
+      headers:
+        Authorization: "Basic dGVzdF91c2VyOndyb25nLXBhc3N3b3Jk" # invalid credentials
+      bulk:
+        body: |
+          {"index": {}}
+          {}
+      catch: unauthorized
+  - match:
+      error.root_cause.0.type: security_exception