Fix Broken HTTP Request Breaking Channel Closing (#45958)

This is essentially the same issue fixed in #43362 but for http request
version instead of the request method. We have to deal with the
case of not being able to parse the request version, otherwise
channel closing fails.

Fixes #43850

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

@@ -168,12 +168,13 @@ public class DefaultRestChannel extends AbstractRestChannel implements RestChann
 
     // Determine if the request connection should be closed on completion.
     private boolean isCloseConnection() {
-        final boolean http10 = isHttp10();
-        return CLOSE.equalsIgnoreCase(request.header(CONNECTION)) || (http10 && !KEEP_ALIVE.equalsIgnoreCase(request.header(CONNECTION)));
-    }
-
-    // Determine if the request protocol version is HTTP 1.0
-    private boolean isHttp10() {
-        return request.getHttpRequest().protocolVersion() == HttpRequest.HttpVersion.HTTP_1_0;
+        try {
+            final boolean http10 = request.getHttpRequest().protocolVersion() == HttpRequest.HttpVersion.HTTP_1_0;
+            return CLOSE.equalsIgnoreCase(request.header(CONNECTION))
+                || (http10 && !KEEP_ALIVE.equalsIgnoreCase(request.header(CONNECTION)));
+        } catch (Exception e) {
+            // In case we fail to parse the http protocol version out of the request we always close the connection
+            return true;
+        }
     }
 }

server/src/test/java/org/elasticsearch/http/DefaultRestChannelTests.java

@@ -57,6 +57,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Supplier;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
@@ -272,8 +273,13 @@ public class DefaultRestChannelTests extends ESTestCase {
     public void testConnectionClose() throws Exception {
         final Settings settings = Settings.builder().build();
         final HttpRequest httpRequest;
-        final boolean close = randomBoolean();
-        if (randomBoolean()) {
+        final boolean brokenRequest = randomBoolean();
+        final boolean close = brokenRequest || randomBoolean();
+        if (brokenRequest) {
+            httpRequest = new TestRequest(() -> {
+                throw new IllegalArgumentException("Can't parse HTTP version");
+            }, RestRequest.Method.GET, "/");
+        } else if (randomBoolean()) {
             httpRequest = new TestRequest(HttpRequest.HttpVersion.HTTP_1_1, RestRequest.Method.GET, "/");
             if (close) {
                 httpRequest.getHeaders().put(DefaultRestChannel.CONNECTION, Collections.singletonList(DefaultRestChannel.CLOSE));
@@ -399,18 +405,21 @@ public class DefaultRestChannelTests extends ESTestCase {
 
     private static class TestRequest implements HttpRequest {
 
-        private final HttpVersion version;
+        private final Supplier<HttpVersion> version;
         private final RestRequest.Method method;
         private final String uri;
         private HashMap<String, List<String>> headers = new HashMap<>();
 
-        private TestRequest(HttpVersion version, RestRequest.Method method, String uri) {
-
-            this.version = version;
+        private TestRequest(Supplier<HttpVersion> versionSupplier, RestRequest.Method method, String uri) {
+            this.version = versionSupplier;
             this.method = method;
             this.uri = uri;
         }
 
+        private TestRequest(HttpVersion version, RestRequest.Method method, String uri) {
+            this(() -> version, method, uri);
+        }
+
         @Override
         public RestRequest.Method method() {
             return method;
@@ -438,7 +447,7 @@ public class DefaultRestChannelTests extends ESTestCase {
 
         @Override
         public HttpVersion protocolVersion() {
-            return version;
+            return version.get();
         }
 
         @Override