Add injection of custom executor service to S3Base supplyAsync calls

api/src/main/java/io/minio/MinioAsyncClient.java

@@ -84,6 +84,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
 import java.util.regex.Matcher;
 import okhttp3.HttpUrl;
 import okhttp3.MultipartBody;
@@ -149,7 +151,8 @@ private MinioAsyncClient(
       String region,
       Provider provider,
       OkHttpClient httpClient,
-      boolean closeHttpClient) {
+      boolean closeHttpClient,
+      ExecutorService executorService) {
     super(
         baseUrl,
         awsS3Prefix,
@@ -159,7 +162,8 @@ private MinioAsyncClient(
         region,
         provider,
         httpClient,
-        closeHttpClient);
+        closeHttpClient,
+        executorService);
   }
 
   protected MinioAsyncClient(MinioAsyncClient client) {
@@ -462,7 +466,7 @@ public CompletableFuture<ObjectWriteResponse> copyObject(CopyObjectArgs args)
     args.validateSse(this.baseUrl);
 
     return CompletableFuture.supplyAsync(
-            () -> args.source().offset() != null && args.source().length() != null)
+            () -> args.source().offset() != null && args.source().length() != null, executorService)
         .thenCompose(
             condition -> {
               if (condition) {
@@ -677,7 +681,7 @@ public CompletableFuture<ObjectWriteResponse> composeObject(ComposeObjectArgs ar
                             Multimap<String, String> headers = newMultimap(args.extraHeaders());
                             headers.putAll(args.genHeaders());
                             return headers;
-                          })
+                          }, executorService)
                       .thenCompose(
                           headers -> {
                             try {
@@ -715,7 +719,7 @@ public CompletableFuture<ObjectWriteResponse> composeObject(ComposeObjectArgs ar
                                 CompletableFuture.supplyAsync(
                                     () -> {
                                       return new Part[partCount[0]];
-                                    });
+                                    }, executorService);
                             for (ComposeSource src : sources) {
                               long size = 0;
                               try {
@@ -3491,7 +3495,7 @@ public CompletableFuture<ObjectWriteResponse> uploadSnowballObjects(
                 }
               }
               return baos;
-            })
+            }, executorService)
         .thenCompose(
             baos -> {
               Multimap<String, String> headers = newMultimap(args.extraHeaders());
@@ -3633,7 +3637,7 @@ public CompletableFuture<PutObjectFanOutResponse> putObjectFanOut(PutObjectFanOu
                   | XmlParserException e) {
                 throw new CompletionException(e);
               }
-            })
+            }, executorService)
         .thenCompose(
             body -> {
               try {
@@ -3721,6 +3725,7 @@ public static final class Builder {
     private Provider provider;
     private OkHttpClient httpClient;
     private boolean closeHttpClient;
+    private ExecutorService executorService = ForkJoinPool.commonPool();
 
     private void setAwsInfo(String host, boolean https) {
       this.awsS3Prefix = null;
@@ -3834,6 +3839,11 @@ public Builder httpClient(OkHttpClient httpClient, boolean close) {
       return this;
     }
 
+    public Builder executorService(ExecutorService executorService) {
+      this.executorService = executorService;
+      return this;
+    }
+
     public MinioAsyncClient build() {
       HttpUtils.validateNotNull(this.baseUrl, "endpoint");
 
@@ -3861,7 +3871,8 @@ public MinioAsyncClient build() {
           region,
           provider,
           httpClient,
-          closeHttpClient);
+          closeHttpClient,
+          executorService);
     }
   }
 }

api/src/main/java/io/minio/S3Base.java

@@ -87,6 +87,8 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
@@ -147,6 +149,7 @@ public abstract class S3Base implements AutoCloseable {
   protected Provider provider;
   protected OkHttpClient httpClient;
   protected boolean closeHttpClient;
+  protected final ExecutorService executorService;
 
   /** @deprecated This method is no longer supported. */
   @Deprecated
@@ -171,6 +174,19 @@ protected S3Base(
         false);
   }
 
+    protected S3Base(
+            HttpUrl baseUrl,
+            String awsS3Prefix,
+            String awsDomainSuffix,
+            boolean awsDualstack,
+            boolean useVirtualStyle,
+            String region,
+            Provider provider,
+            OkHttpClient httpClient,
+            boolean closeHttpClient) {
+        this(baseUrl, awsS3Prefix, awsDomainSuffix, awsDualstack, useVirtualStyle, region, provider, httpClient, closeHttpClient, ForkJoinPool.commonPool());
+    }
+
   protected S3Base(
       HttpUrl baseUrl,
       String awsS3Prefix,
@@ -180,7 +196,8 @@ protected S3Base(
       String region,
       Provider provider,
       OkHttpClient httpClient,
-      boolean closeHttpClient) {
+      boolean closeHttpClient,
+      ExecutorService executorService) {
     this.baseUrl = baseUrl;
     this.awsS3Prefix = awsS3Prefix;
     this.awsDomainSuffix = awsDomainSuffix;
@@ -190,6 +207,7 @@ protected S3Base(
     this.provider = provider;
     this.httpClient = httpClient;
     this.closeHttpClient = closeHttpClient;
+    this.executorService = executorService;
   }
 
   /** @deprecated This method is no longer supported. */
@@ -219,6 +237,7 @@ protected S3Base(
     this.provider = provider;
     this.httpClient = httpClient;
     this.closeHttpClient = false;
+    this.executorService = ForkJoinPool.commonPool();
   }
 
   protected S3Base(S3Base client) {
@@ -231,6 +250,7 @@ protected S3Base(S3Base client) {
     this.provider = client.provider;
     this.httpClient = client.httpClient;
     this.closeHttpClient = client.closeHttpClient;
+    this.executorService = client.executorService;
   }
 
   /** Check whether argument is valid or not. */
@@ -1204,7 +1224,8 @@ protected CompletableFuture<Integer> calculatePartCountAsync(List<ComposeSource>
     long[] objectSize = {0};
     int index = 0;
 
-    CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> 0);
+    CompletableFuture<Integer> completableFuture =
+        CompletableFuture.supplyAsync(() -> 0, executorService);
     for (ComposeSource src : sources) {
       index++;
       final int i = index;
@@ -2947,7 +2968,8 @@ private CompletableFuture<ObjectWriteResponse> putMultipartObjectAsync(
             throw new CompletionException(throwable);
           }
           return response;
-        });
+        },
+        executorService);
   }
 
   /**
@@ -2993,7 +3015,8 @@ protected CompletableFuture<ObjectWriteResponse> putObjectAsync(
               } catch (NoSuchAlgorithmException | IOException e) {
                 throw new CompletionException(e);
               }
-            })
+            },
+            executorService)
         .thenCompose(
             partSource -> {
               try {