Fix Broken HTTP Request Breaking Channel Closing (elastic#45958)

This is essentially the same issue fixed in elastic#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 elastic#43850

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

@@ -168,12 +168,13 @@ private void addCookies(HttpResponse response) {
 
     // 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.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 void testReleaseInListener() throws IOException {
     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 @@ private TestResponse executeRequest(final Settings settings, final String origin
 
     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 List<String> strictCookies() {
 
         @Override
         public HttpVersion protocolVersion() {
-            return version;
+            return version.get();
         }
 
         @Override