Allow consumers to provide their executor service (#637)

Signed-off-by: Valentin Delaye <jonesbusy@users.noreply.github.com>

Author: jonesbusy

src/main/java/land/oras/Registry.java

@@ -66,7 +66,7 @@ public final class Registry extends OCI<ContainerRef> {
     /**
      * The executor service for parallel operations
      */
-    private ExecutorService executors;
+    private ExecutorService executorService;
 
     /**
      * The registries configuration loaded from the environment
@@ -139,6 +139,14 @@ private void setParallelism(int maxConcurrentDownloads) {
         this.maxConcurrentDownloads = maxConcurrentDownloads;
     }
 
+    /**
+     * Allow consumer to set custom executor service for parallel operations. If not set, a default one will be created with the given parallelism
+     * @param executorService The executor service
+     */
+    private void setExecutorService(ExecutorService executorService) {
+        this.executorService = executorService;
+    }
+
     /**
      * Return if this registry is insecure
      * @return True if insecure
@@ -177,11 +185,13 @@ private void setSkipTlsVerify(boolean skipTlsVerify) {
      */
     private Registry build() {
         client = HttpClient.Builder.builder().withSkipTlsVerify(skipTlsVerify).build();
-        executors = Executors.newFixedThreadPool(maxConcurrentDownloads, r -> {
-            Thread t = new Thread(r);
-            t.setName("layer-transfer-worker-%d".formatted(t.getId()));
-            return t;
-        });
+        if (executorService == null) {
+            executorService = Executors.newFixedThreadPool(maxConcurrentDownloads, r -> {
+                Thread t = new Thread(r);
+                t.setName("layer-transfer-worker-%d".formatted(t.getId()));
+                return t;
+            });
+        }
         return this;
     }
 
@@ -220,7 +230,7 @@ public Registry asInsecure() {
 
     @Override
     public ExecutorService getExecutorService() {
-        return executors;
+        return executorService;
     }
 
     @Override
@@ -1108,6 +1118,7 @@ public Builder from(Registry registry) {
             this.registry.setInsecure(registry.insecure);
             this.registry.setRegistry(registry.registry);
             this.registry.setSkipTlsVerify(registry.skipTlsVerify);
+            this.registry.setExecutorService(registry.executorService);
             this.registry.setParallelism(registry.maxConcurrentDownloads);
             return this;
         }
@@ -1195,6 +1206,17 @@ public Builder withParallelism(int parallelism) {
             return this;
         }
 
+        /**
+         * Set the executor service to use for parallel uploads/downloads. By default it uses a parallelism level given by withParallelism() and a fixed thread pool.
+         * Only uses for layers upload/download, not for manifest or index upload/download.
+         * @param executorService The executor service
+         * @return The builder
+         */
+        public Builder withExecutorService(ExecutorService executorService) {
+            registry.setExecutorService(executorService);
+            return this;
+        }
+
         /**
          * Set the insecure flag
          * @param insecure Insecure

src/test/java/land/oras/DockerIoITCase.java

@@ -107,12 +107,12 @@ void shouldCopyTagToInternalRegistry() {
 
         // Source registry
         Registry sourceRegistry =
-                Registry.Builder.builder().withParallelism(10).defaults().build();
+                Registry.Builder.builder().withParallelism(3).defaults().build();
 
         // Copy to this internal registry
         Registry targetRegistry = Registry.Builder.builder()
                 .defaults("myuser", "mypass")
-                .withParallelism(10)
+                .withParallelism(3)
                 .withInsecure(true)
                 .build();
 

src/test/java/land/oras/RegistryTest.java

@@ -32,6 +32,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import land.oras.exception.OrasException;
 import land.oras.utils.Const;
 import land.oras.utils.RegistryContainer;
@@ -325,6 +327,8 @@ void shouldPushUnsecure() {
     @Execution(ExecutionMode.SAME_THREAD)
     void shouldPushPullManifestsAndBlobsByUsingConfig(@TempDir Path homeDir) throws Exception {
 
+        ExecutorService customExecutor = Executors.newSingleThreadExecutor();
+
         // language=toml
         String config =
                 """
@@ -336,7 +340,9 @@ void shouldPushPullManifestsAndBlobsByUsingConfig(@TempDir Path homeDir) throws
         TestUtils.createRegistriesConfFile(homeDir, config);
 
         TestUtils.withHome(homeDir, () -> {
-            Registry registry = Registry.Builder.builder().build(); // Use default
+            Registry registry = Registry.Builder.builder()
+                    .withExecutorService(customExecutor)
+                    .build();
             ContainerRef containerRef = ContainerRef.parse(
                     "%s/library/artifact-text-manifest-blobs".formatted(this.unsecureRegistry.getRegistry()));
 
@@ -370,6 +376,8 @@ void shouldPushPullManifestsAndBlobsByUsingConfig(@TempDir Path homeDir) throws
             registry.deleteManifest(containerRef);
             registry.deleteBlob(containerRef.withDigest(otherDigest));
         });
+
+        customExecutor.shutdown();
     }
 
     @Test