Fix RemoteFileUtil to download in parallel as expected.

scio-core/src/main/java/com/spotify/scio/util/RemoteFileUtil.java

@@ -29,7 +29,11 @@
 import java.nio.file.Paths;
 import java.nio.file.StandardOpenOption;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
 import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
 import org.apache.beam.sdk.io.fs.ResourceId;
@@ -103,10 +107,37 @@ public Path download(URI src) {
    * @return {@link Path}s to the downloaded local files.
    */
   public List<Path> download(List<URI> srcs) {
+    return download(srcs, CONCURRENCY_LEVEL);
+  }
+
+  /**
+   * Download a batch of remote {@link URI}s in parallel, using at most numThreads to do so.
+   * `numThreads` may not be larger than the number of available processors * 4.
+   *
+   * @return {@link Path}s to the downloaded local files.
+   */
+  public List<Path> download(List<URI> srcs, int numThreads) {
+    final ExecutorService executor = Executors.newFixedThreadPool(numThreads);
     try {
-      return paths.getAll(srcs).values().asList();
-    } catch (ExecutionException e) {
+      return executor
+          .invokeAll(
+              srcs.stream()
+                  .map(url -> (Callable<Path>) () -> paths.get(url))
+                  .collect(Collectors.toList()))
+          .stream()
+          .map(
+              f -> {
+                try {
+                  return f.get();
+                } catch (InterruptedException | ExecutionException e) {
+                  throw new RuntimeException(e);
+                }
+              })
+          .collect(Collectors.toList());
+    } catch (InterruptedException e) {
       throw new RuntimeException(e);
+    } finally {
+      executor.shutdown();
     }
   }