Fix missing flush in testReadBlobWithPrematureConnectionClose (#114094)

* Fix missing flush in `testReadBlobWithPrematureConnectionClose`

It seems that JDK23 yields a different error message if the connection
is closed before even sending headers. This commit relaxes the assertion
to match, and also tightens it up to check that we report a premature
end of the body if we _do_ send headers.

Backport of #113609 to `7.x`
Closes #114016

* Fix license header

Author: DaveCTurner

test/framework/src/main/java/org/elasticsearch/repositories/blobstore/AbstractBlobContainerRetriesTestCase.java

@@ -41,6 +41,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import static org.elasticsearch.test.NeverMatcher.never;
+import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.either;
 import static org.hamcrest.Matchers.equalTo;
@@ -277,9 +279,12 @@ public void testReadBlobWithReadTimeouts() {
         assertThat(exception, readTimeoutExceptionMatcher());
         assertThat(
             exception.getMessage().toLowerCase(Locale.ROOT),
-            either(containsString("read timed out")).or(containsString("premature end of chunk coded message body: closing chunk expected"))
-                .or(containsString("Read timed out"))
-                .or(containsString("unexpected end of file from server"))
+            anyOf(
+                containsString("read timed out"),
+                containsString("premature end of chunk coded message body: closing chunk expected"),
+                containsString("Read timed out"),
+                containsString("unexpected end of file from server")
+            )
         );
         assertThat(exception.getSuppressed().length, equalTo(maxRetries));
     }
@@ -308,10 +313,15 @@ public void testReadBlobWithPrematureConnectionClose() {
         final int maxRetries = randomInt(20);
         final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null);
 
+        final boolean alwaysFlushBody = randomBoolean();
+
         // HTTP server sends a partial response
         final byte[] bytes = randomBlobContent(1);
         httpServer.createContext(downloadStorageEndpoint(blobContainer, "read_blob_incomplete"), exchange -> {
             sendIncompleteContent(exchange, bytes);
+            if (alwaysFlushBody) {
+                exchange.getResponseBody().flush();
+            }
             exchange.close();
         });
 
@@ -326,9 +336,14 @@ public void testReadBlobWithPrematureConnectionClose() {
         });
         assertThat(
             exception.getMessage().toLowerCase(Locale.ROOT),
-            either(containsString("premature end of chunk coded message body: closing chunk expected")).or(
-                containsString("premature end of content-length delimited message body")
-            ).or(containsString("connection closed prematurely"))
+            anyOf(
+                // closing the connection after sending the headers and some incomplete body might yield one of these:
+                containsString("premature end of chunk coded message body: closing chunk expected"),
+                containsString("premature end of content-length delimited message body"),
+                containsString("connection closed prematurely"),
+                // if we didn't call exchange.getResponseBody().flush() then we might not even have sent the response headers:
+                alwaysFlushBody ? never() : containsString("the target server failed to respond")
+            )
         );
         assertThat(exception.getSuppressed().length, equalTo(Math.min(10, maxRetries)));
     }

test/framework/src/main/java/org/elasticsearch/test/NeverMatcher.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+
+public class NeverMatcher<T> extends BaseMatcher<T> {
+    @SuppressWarnings("unchecked")
+    public static <T> Matcher<T> never() {
+        return (Matcher<T>) INSTANCE;
+    }
+
+    private static final Matcher<?> INSTANCE = new NeverMatcher<>();
+
+    private NeverMatcher() {/* singleton */}
+
+    @Override
+    public boolean matches(Object actual) {
+        return false;
+    }
+
+    @Override
+    public void describeTo(Description description) {
+        description.appendText("never matches");
+    }
+}