Test max connections

Author: DiannaHohensee

modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainerRetriesTests.java

@@ -126,6 +126,7 @@ protected BlobContainer createBlobContainer(
         final @Nullable Integer maxRetries,
         final @Nullable TimeValue readTimeout,
         final @Nullable Boolean disableChunkedEncoding,
+        final @Nullable Integer maxConnections,
         final @Nullable ByteSizeValue bufferSize,
         final @Nullable Integer maxBulkDeletes,
         BlobPath blobContainerPath
@@ -225,7 +226,7 @@ public void testShouldRetryOnUnresolvableHost() {
 
     private void executeListBlobsAndAssertRetries() {
         final int maxRetries = randomIntBetween(3, 5);
-        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null, null);
         expectThrows(StorageException.class, () -> blobContainer.listBlobs(randomPurpose()));
         assertEquals(maxRetries + 1, requestCounters.get("/storage/v1/b/bucket/o").get());
     }
@@ -234,7 +235,7 @@ public void testReadLargeBlobWithRetries() throws Exception {
         final int maxRetries = randomIntBetween(2, 10);
         final AtomicInteger countDown = new AtomicInteger(maxRetries);
 
-        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null, null);
 
         // SDK reads in 2 MB chunks so we use twice that to simulate 2 chunks
         final byte[] bytes = randomBytes(1 << 22);
@@ -263,7 +264,7 @@ public void testWriteBlobWithRetries() throws Exception {
         final int maxRetries = randomIntBetween(2, 10);
         final CountDown countDown = new CountDown(maxRetries);
 
-        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null, null, null);
         final byte[] bytes = randomBlobContent();
         httpServer.createContext("/upload/storage/v1/b/bucket/o", safeHandler(exchange -> {
             assertThat(exchange.getRequestURI().getQuery(), containsString("uploadType=multipart"));
@@ -305,7 +306,7 @@ public void testWriteBlobWithRetries() throws Exception {
     public void testWriteBlobWithReadTimeouts() {
         final byte[] bytes = randomByteArrayOfLength(randomIntBetween(10, 128));
         final TimeValue readTimeout = TimeValue.timeValueMillis(randomIntBetween(100, 500));
-        final BlobContainer blobContainer = createBlobContainer(1, readTimeout, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(1, readTimeout, null, null, null, null, null);
 
         // HTTP server does not send a response
         httpServer.createContext("/upload/storage/v1/b/bucket/o", exchange -> {
@@ -357,7 +358,7 @@ public void testWriteLargeBlob() throws IOException {
         logger.debug("starting with resumable upload id [{}]", sessionUploadId.get());
 
         final TimeValue readTimeout = allowReadTimeout.get() ? TimeValue.timeValueSeconds(3) : null;
-        final BlobContainer blobContainer = createBlobContainer(nbErrors + 1, readTimeout, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(nbErrors + 1, readTimeout, null, null, null, null, null);
 
         httpServer.createContext("/upload/storage/v1/b/bucket/o", safeHandler(exchange -> {
             final BytesReference requestBody = Streams.readFully(exchange.getRequestBody());
@@ -504,7 +505,7 @@ public String next() {
                 return Integer.toString(totalDeletesSent++);
             }
         };
-        final BlobContainer blobContainer = createBlobContainer(1, null, null, null, null, null);
+        final BlobContainer blobContainer = createBlobContainer(1, null, null, null, null, null, null);
         httpServer.createContext("/batch/storage/v1", safeHandler(exchange -> {
             assert pendingDeletes.get() <= MAX_DELETES_PER_BATCH;
 
@@ -540,7 +541,7 @@ public void testCompareAndExchangeWhenThrottled() throws IOException {
         httpServer.createContext("/", new ResponseInjectingHttpHandler(requestHandlers, new GoogleCloudStorageHttpHandler("bucket")));
 
         final int maxRetries = randomIntBetween(1, 3);
-        final BlobContainer container = createBlobContainer(maxRetries, null, null, null, null, null);
+        final BlobContainer container = createBlobContainer(maxRetries, null, null, null, null, null, null);
         final byte[] data = randomBytes(randomIntBetween(1, BlobContainerUtils.MAX_REGISTER_CONTENT_LENGTH));
         final String key = randomIdentifier();
 

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -292,10 +292,25 @@ DnsResolver getCustomDnsResolver() {
         return null;
     }
 
-    static SdkHttpClient buildHttpClient(S3ClientSettings clientSettings, @Nullable /* to use default resolver */ DnsResolver dnsResolver) {
+    /**
+     * An override for testing purposes.
+     */
+    Optional<Duration> getConnectionAcquisitionTimeout() {
+        return Optional.empty();
+    }
+
+    private SdkHttpClient buildHttpClient(
+        S3ClientSettings clientSettings,
+        @Nullable /* to use default resolver */ DnsResolver dnsResolver
+    ) {
         ApacheHttpClient.Builder httpClientBuilder = ApacheHttpClient.builder();
 
-        // TODO NOMERGE: an IT test for maxConnections?
+        var optConnectionAcquisitionTimout = getConnectionAcquisitionTimeout();
+        if (optConnectionAcquisitionTimout.isPresent()) {
+            // Only tests set this.
+            httpClientBuilder.connectionAcquisitionTimeout(optConnectionAcquisitionTimout.get());
+        }
+
         httpClientBuilder.maxConnections(clientSettings.maxConnections);
         httpClientBuilder.socketTimeout(Duration.ofMillis(clientSettings.readTimeoutMillis));