Reduce allocations when draining HTTP requests bodies in repository tests (#48541)

In repository integration tests, we drain the HTTP request body before 
returning a response. Before this change this operation was done using
Streams.readFully() which uses a 8kb buffer to read the input stream, it
 now uses a 1kb for the same operation. This should reduce the allocations 
made during the tests and speed them up a bit on CI.

Co-authored-by: Armin Braun <me@obrown.io>

plugins/repository-azure/src/test/java/org/elasticsearch/repositories/azure/AzureBlobStoreRepositoryTests.java

@@ -199,7 +199,7 @@ public class AzureBlobStoreRepositoryTests extends ESMockAPIBasedRepositoryInteg
                     exchange.getResponseBody().write(blob.toBytesRef().bytes, start, length);
 
                 } else if (Regex.simpleMatch("DELETE /container/*", request)) {
-                    Streams.readFully(exchange.getRequestBody());
+                    drainInputStream(exchange.getRequestBody());
                     blobs.entrySet().removeIf(blob -> blob.getKey().startsWith(exchange.getRequestURI().getPath()));
                     exchange.sendResponseHeaders(RestStatus.ACCEPTED.getStatus(), -1);
 
@@ -251,7 +251,7 @@ public class AzureBlobStoreRepositoryTests extends ESMockAPIBasedRepositoryInteg
 
         @Override
         protected void handleAsError(final HttpExchange exchange) throws IOException {
-            Streams.readFully(exchange.getRequestBody());
+            drainInputStream(exchange.getRequestBody());
             TestUtils.sendError(exchange, randomFrom(RestStatus.INTERNAL_SERVER_ERROR, RestStatus.SERVICE_UNAVAILABLE));
             exchange.close();
         }

plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -194,7 +194,7 @@ public class S3BlobStoreRepositoryTests extends ESMockAPIBasedRepositoryIntegTes
                     }
 
                 } else if (Regex.simpleMatch("POST /bucket/*?uploadId=*", request)) {
-                    Streams.readFully(exchange.getRequestBody());
+                    drainInputStream(exchange.getRequestBody());
                     final Map<String, String> params = new HashMap<>();
                     RestUtils.decodeQueryString(exchange.getRequestURI().getQuery(), 0, params);
                     final String uploadId = params.get("uploadId");

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

@@ -26,7 +26,6 @@ import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.SuppressForbidden;
-import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.mocksocket.MockHttpServer;
@@ -37,6 +36,7 @@ import org.junit.Before;
 import org.junit.BeforeClass;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Map;
@@ -53,6 +53,8 @@ import static org.hamcrest.Matchers.equalTo;
 @SuppressForbidden(reason = "this test uses a HttpServer to emulate a cloud-based storage service")
 public abstract class ESMockAPIBasedRepositoryIntegTestCase extends ESBlobStoreRepositoryIntegTestCase {
 
+    private static final byte[] BUFFER = new byte[1024];
+
     private static HttpServer httpServer;
     private Map<String, HttpHandler> handlers;
 
@@ -127,6 +129,15 @@ public abstract class ESMockAPIBasedRepositoryIntegTestCase extends ESBlobStoreR
         return "http://" + InetAddresses.toUriString(address.getAddress()) + ":" + address.getPort();
     }
 
+    /**
+     * Consumes and closes the given {@link InputStream}
+     */
+    protected static void drainInputStream(final InputStream inputStream) throws IOException {
+        try (InputStream is = inputStream) {
+            while (is.read(BUFFER) >= 0);
+        }
+    }
+
     /**
      * HTTP handler that injects random service errors
      *
@@ -166,7 +177,7 @@ public abstract class ESMockAPIBasedRepositoryIntegTestCase extends ESBlobStoreR
         }
 
         protected void handleAsError(final HttpExchange exchange) throws IOException {
-            Streams.readFully(exchange.getRequestBody());
+            drainInputStream(exchange.getRequestBody());
             exchange.sendResponseHeaders(HttpStatus.SC_INTERNAL_SERVER_ERROR, -1);
             exchange.close();
         }