Repository: googleapis/java-storage
Pull request #1990 (2023-04-20 16:44:18)

feat: Implement chunkedDownload and refactor direct downloads.
Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-storage/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ChunkedDownloadCallable.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ChunkedDownloadCallable.java
new file mode 100644
index 00000000..86e2a9cd
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ChunkedDownloadCallable.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.StorageException;
+import com.google.common.io.ByteStreams;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
+
+final class ChunkedDownloadCallable implements Callable<DownloadSegment> {
+
+  private final BlobInfo originalBlob;
+
+  private final Storage storage;
+
+  private final Storage.BlobSourceOption[] opts;
+
+  private final long startPosition;
+
+  private final long endPosition;
+  private final Path destPath;
+
+  ChunkedDownloadCallable(
+      Storage storage,
+      BlobInfo originalBlob,
+      BlobSourceOption[] opts,
+      Path destPath,
+      long startPosition,
+      long endPosition) {
+    this.originalBlob = originalBlob;
+    this.storage = storage;
+    this.opts = opts;
+    this.startPosition = startPosition;
+    this.endPosition = endPosition;
+    this.destPath = destPath;
+  }
+
+  @Override
+  public DownloadSegment call() throws Exception {
+    // TODO: Wrap Exceptions and put them in result
+    try (ReadChannel rc = storage.reader(originalBlob.getBlobId(), opts)) {
+      FileChannel wc =
+          FileChannel.open(destPath, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
+      rc.seek(startPosition);
+      rc.limit(endPosition);
+      wc.position(startPosition);
+      ByteStreams.copy(rc, wc);
+    } catch (IOException e) {
+      throw new StorageException(e);
+    }
+    DownloadSegment result =
+        DownloadSegment.newBuilder(originalBlob, TransferStatus.SUCCESS)
+            .setOutputDestination(destPath)
+            .build();
+    return result;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java
new file mode 100644
index 00000000..fa1ed0c9
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+final class DefaultQos implements Qos {
+
+  private final long divideAndConquerThreshold;
+
+  private DefaultQos(long divideAndConquerThreshold) {
+    this.divideAndConquerThreshold = divideAndConquerThreshold;
+  }
+
+  @Override
+  public boolean divideAndConquer(long objectSize) {
+    return objectSize > divideAndConquerThreshold;
+  }
+
+  static DefaultQos of() {
+    return of(128L * 1024 * 1024);
+  }
+
+  static DefaultQos of(long divideAndConquerThreshold) {
+    return new DefaultQos(divideAndConquerThreshold);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadCallable.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java
similarity index 66%
rename from google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadCallable.java
rename to google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java
index db6640ea..cb008c05 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadCallable.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java
@@ -24,12 +24,11 @@ import com.google.cloud.storage.StorageException;
 import com.google.common.io.ByteStreams;
 import java.io.IOException;
 import java.nio.channels.FileChannel;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.Callable;
 
-final class DownloadCallable implements Callable<DownloadResult> {
+final class DirectDownloadCallable implements Callable<DownloadResult> {
   private final BlobInfo originalBlob;
 
   private final ParallelDownloadConfig parallelDownloadConfig;
@@ -37,7 +36,7 @@ final class DownloadCallable implements Callable<DownloadResult> {
 
   private final Storage.BlobSourceOption[] opts;
 
-  DownloadCallable(
+  DirectDownloadCallable(
       Storage storage,
       BlobInfo originalBlob,
       ParallelDownloadConfig parallelDownloadConfig,
@@ -49,48 +48,23 @@ final class DownloadCallable implements Callable<DownloadResult> {
   }
 
   @Override
-  public DownloadResult call() throws Exception {
-    // TODO: Check for chunking
-    return downloadWithoutChunking();
-  }
-
-  private DownloadResult downloadWithoutChunking() {
-    Path path = createDestPath();
+  public DownloadResult call() {
+    Path path = TransferManagerUtils.createDestPath(parallelDownloadConfig, originalBlob);
     try (ReadChannel rc = storage.reader(originalBlob.getBlobId(), opts)) {
-      FileChannel destFile =
+      FileChannel wc =
           FileChannel.open(
               path,
               StandardOpenOption.WRITE,
               StandardOpenOption.CREATE,
               StandardOpenOption.TRUNCATE_EXISTING);
-      ByteStreams.copy(rc, destFile);
+      ByteStreams.copy(rc, wc);
     } catch (IOException e) {
       throw new StorageException(e);
     }
     DownloadResult result =
         DownloadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
-            .setOutputDestination(path)
+            .setOutputDestination(path.toAbsolutePath())
             .build();
     return result;
   }
-
-  private Path createDestPath() {
-    Path newPath =
-        parallelDownloadConfig
-            .getDownloadDirectory()
-            .resolve(
-                originalBlob.getName().replaceFirst(parallelDownloadConfig.getStripPrefix(), ""));
-    // Check to make sure the parent directories exist
-    if (Files.exists(newPath.getParent())) {
-      return newPath;
-    } else {
-      // Make parent directories if they do not exist
-      try {
-        Files.createDirectories(newPath.getParent());
-        return newPath;
-      } catch (IOException e) {
-        throw new StorageException(e);
-      }
-    }
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
index af2d8c70..db427d8b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
@@ -33,17 +33,17 @@ public final class DownloadJob {
   @NonNull private final ParallelDownloadConfig parallelDownloadConfig;
 
   private DownloadJob(
-      @NonNull List<ApiFuture<DownloadResult>> successResponses,
+      @NonNull List<ApiFuture<DownloadResult>> downloadResults,
       @NonNull ParallelDownloadConfig parallelDownloadConfig) {
-    this.downloadResults = successResponses;
+    this.downloadResults = downloadResults;
     this.parallelDownloadConfig = parallelDownloadConfig;
   }
 
-  public List<ApiFuture<DownloadResult>> getDownloadResults() {
+  public @NonNull List<ApiFuture<DownloadResult>> getDownloadResults() {
     return downloadResults;
   }
 
-  public ParallelDownloadConfig getParallelDownloadConfig() {
+  public @NonNull ParallelDownloadConfig getParallelDownloadConfig() {
     return parallelDownloadConfig;
   }
 
@@ -68,7 +68,7 @@ public final class DownloadJob {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("successResponses", downloadResults)
+        .add("downloadResults", downloadResults)
         .add("parallelDownloadConfig", parallelDownloadConfig)
         .toString();
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
index 27e4c952..143154d7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
@@ -23,11 +23,14 @@ import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.StorageException;
 import com.google.common.base.MoreObjects;
 import java.nio.file.Path;
+import java.util.Comparator;
 import java.util.Objects;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 public final class DownloadResult {
+  static final Comparator<DownloadResult> COMPARATOR =
+      Comparator.comparingInt(dr -> dr.getStatus().ordinal());
 
   @NonNull private final BlobInfo input;
   @MonotonicNonNull private final Path outputDestination;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadSegment.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadSegment.java
new file mode 100644
index 00000000..24a83514
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadSegment.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.StorageException;
+import java.nio.file.Path;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class DownloadSegment {
+  private final BlobInfo input;
+
+  private final Path outputDestination;
+
+  private final TransferStatus status;
+
+  private final StorageException exception;
+
+  private final Long generation;
+
+  private DownloadSegment(
+      BlobInfo input,
+      Path outputDestination,
+      TransferStatus status,
+      StorageException exception,
+      Long generation) {
+    this.input = input;
+    this.outputDestination = outputDestination;
+    this.status = status;
+    this.exception = exception;
+    this.generation = generation;
+  }
+
+  public BlobInfo getInput() {
+    return input;
+  }
+
+  public Path getOutputDestination() {
+    return outputDestination;
+  }
+
+  public TransferStatus getStatus() {
+    return status;
+  }
+
+  public StorageException getException() {
+    return exception;
+  }
+
+  public static Builder newBuilder(BlobInfo input, TransferStatus status) {
+    return new Builder(input, status);
+  }
+
+  public Long getGeneration() {
+    return generation;
+  }
+
+  public DownloadResult toResult() {
+    DownloadResult.Builder b = DownloadResult.newBuilder(input, status);
+    if (exception != null) {
+      b.setException(exception);
+    }
+    if (outputDestination != null) {
+      b.setOutputDestination(outputDestination);
+    }
+    return b.build();
+  }
+
+  @NonNull
+  public static DownloadResult reduce(
+      @NonNull DownloadResult result, @NonNull DownloadSegment segment) {
+    if (TransferStatus.COMPARE_NULL_SAFE.compare(segment.getStatus(), result.getStatus()) <= 0) {
+      return result;
+    } else {
+      return segment.toResult();
+    }
+  }
+
+  public static final class Builder {
+
+    private BlobInfo input;
+    private Path outputDestination;
+    private TransferStatus status;
+    private StorageException exception;
+    private Long generation;
+
+    private Builder(BlobInfo input, TransferStatus status) {
+      this.input = input;
+      this.status = status;
+    }
+
+    public Builder setInput(BlobInfo input) {
+      this.input = input;
+      return this;
+    }
+
+    public Builder setOutputDestination(Path outputDestination) {
+      this.outputDestination = outputDestination;
+      return this;
+    }
+
+    public Builder setStatus(TransferStatus status) {
+      this.status = status;
+      return this;
+    }
+
+    public Builder setException(StorageException exception) {
+      this.exception = exception;
+      return this;
+    }
+
+    public Builder setGeneration(Long generation) {
+      this.generation = generation;
+      return this;
+    }
+
+    public DownloadSegment build() {
+      return new DownloadSegment(input, outputDestination, status, exception, generation);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java
new file mode 100644
index 00000000..10059164
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+interface Qos {
+
+  boolean divideAndConquer(long objectSize);
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
index bc250e6e..d55f44da 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
@@ -22,7 +22,7 @@ import java.nio.file.Path;
 import java.util.List;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
-public interface TransferManager {
+public interface TransferManager extends AutoCloseable {
 
   @NonNull
   UploadJob uploadFiles(List<Path> files, ParallelUploadConfig opts) throws IOException;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
index 086f4f1c..ff773583 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
@@ -23,19 +23,22 @@ import java.util.Objects;
 public final class TransferManagerConfig {
   private final int maxWorkers;
   private final int perWorkerBufferSize;
-  private final boolean allowChunking;
+  private final boolean allowDivideAndConquer;
 
   private final StorageOptions storageOptions;
+  private final Qos qos;
 
   TransferManagerConfig(
       int maxWorkers,
       int perWorkerBufferSize,
-      boolean allowChunking,
-      StorageOptions storageOptions) {
+      boolean allowDivideAndConquer,
+      StorageOptions storageOptions,
+      Qos qos) {
     this.maxWorkers = maxWorkers;
     this.perWorkerBufferSize = perWorkerBufferSize;
-    this.allowChunking = allowChunking;
+    this.allowDivideAndConquer = allowDivideAndConquer;
     this.storageOptions = storageOptions;
+    this.qos = qos;
   }
 
   /** Maximum amount of workers to be allocated to perform work in Transfer Manager */
@@ -49,11 +52,11 @@ public final class TransferManagerConfig {
   }
 
   /**
-   * Whether to allow Transfer Manager to performing chunked Uploads/Downloads if it determines
+   * Whether to allow Transfer Manager to perform chunked Uploads/Downloads if it determines
    * chunking will be beneficial
    */
-  public boolean isAllowChunking() {
-    return allowChunking;
+  public boolean isAllowDivideAndConquer() {
+    return allowDivideAndConquer;
   }
 
   /** Storage options that Transfer Manager will use to interact with GCS */
@@ -65,6 +68,19 @@ public final class TransferManagerConfig {
     return new TransferManagerImpl(this);
   }
 
+  public Builder toBuilder() {
+    return new Builder()
+        .setAllowDivideAndConquer(allowDivideAndConquer)
+        .setMaxWorkers(maxWorkers)
+        .setPerWorkerBufferSize(perWorkerBufferSize)
+        .setQos(qos)
+        .setStorageOptions(storageOptions);
+  }
+
+  Qos getQos() {
+    return qos;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -76,13 +92,13 @@ public final class TransferManagerConfig {
     TransferManagerConfig that = (TransferManagerConfig) o;
     return maxWorkers == that.maxWorkers
         && perWorkerBufferSize == that.perWorkerBufferSize
-        && allowChunking == that.allowChunking
+        && allowDivideAndConquer == that.allowDivideAndConquer
         && Objects.equals(storageOptions, that.storageOptions);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(maxWorkers, perWorkerBufferSize, allowChunking, storageOptions);
+    return Objects.hash(maxWorkers, perWorkerBufferSize, allowDivideAndConquer, storageOptions);
   }
 
   @Override
@@ -90,7 +106,7 @@ public final class TransferManagerConfig {
     return MoreObjects.toStringHelper(this)
         .add("maxWorkers", maxWorkers)
         .add("perWorkerBufferSize", perWorkerBufferSize)
-        .add("allowChunking", allowChunking)
+        .add("allowChunking", allowDivideAndConquer)
         .add("storageOptions", storageOptions)
         .toString();
   }
@@ -103,15 +119,17 @@ public final class TransferManagerConfig {
 
     private int maxWorkers;
     private int perWorkerBufferSize;
-    private boolean allowChunking;
+    private boolean allowDivideAndConquer;
 
     private StorageOptions storageOptions;
+    private Qos qos;
 
     private Builder() {
       this.perWorkerBufferSize = 16 * 1024 * 1024;
       this.maxWorkers = 2 * Runtime.getRuntime().availableProcessors();
-      this.allowChunking = false;
+      this.allowDivideAndConquer = false;
       this.storageOptions = StorageOptions.getDefaultInstance();
+      this.qos = DefaultQos.of();
     }
 
     public Builder setMaxWorkers(int maxWorkers) {
@@ -124,18 +142,24 @@ public final class TransferManagerConfig {
       return this;
     }
 
-    public Builder setAllowChunking(boolean allowChunking) {
-      this.allowChunking = allowChunking;
+    public Builder setAllowDivideAndConquer(boolean allowDivideAndConquer) {
+      this.allowDivideAndConquer = allowDivideAndConquer;
       return this;
     }
 
-    public void setStorageOptions(StorageOptions storageOptions) {
+    public Builder setStorageOptions(StorageOptions storageOptions) {
       this.storageOptions = storageOptions;
+      return this;
+    }
+
+    Builder setQos(Qos qos) {
+      this.qos = qos;
+      return this;
     }
 
     public TransferManagerConfig build() {
       return new TransferManagerConfig(
-          maxWorkers, perWorkerBufferSize, allowChunking, storageOptions);
+          maxWorkers, perWorkerBufferSize, allowDivideAndConquer, storageOptions, qos);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
index a9064622..c6c0321c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
@@ -17,7 +17,9 @@
 package com.google.cloud.storage.transfermanager;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
 import com.google.api.core.ListenableFutureToApiFuture;
+import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
@@ -30,12 +32,15 @@ import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BinaryOperator;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class TransferManagerImpl implements TransferManager {
 
   private final TransferManagerConfig transferManagerConfig;
   private final ListeningExecutorService executor;
+  private final Qos qos;
   private final Storage storage;
 
   TransferManagerImpl(TransferManagerConfig transferManagerConfig) {
@@ -43,10 +48,17 @@ final class TransferManagerImpl implements TransferManager {
     this.executor =
         MoreExecutors.listeningDecorator(
             Executors.newFixedThreadPool(transferManagerConfig.getMaxWorkers()));
-
+    this.qos = transferManagerConfig.getQos();
     this.storage = transferManagerConfig.getStorageOptions().getService();
   }
 
+  @Override
+  public void close() throws Exception {
+    // We only want to shutdown the executor service not the provided storage instance
+    executor.shutdownNow();
+    executor.awaitTermination(5, TimeUnit.MINUTES);
+  }
+
   @Override
   public @NonNull UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config) {
     Storage.BlobWriteOption[] opts =
@@ -71,9 +83,48 @@ final class TransferManagerImpl implements TransferManager {
     Storage.BlobSourceOption[] opts =
         config.getOptionsPerRequest().toArray(new Storage.BlobSourceOption[0]);
     List<ApiFuture<DownloadResult>> downloadTasks = new ArrayList<>();
-    for (BlobInfo blob : blobs) {
-      DownloadCallable callable = new DownloadCallable(storage, blob, config, opts);
-      downloadTasks.add(convert(executor.submit(callable)));
+    if (!transferManagerConfig.isAllowDivideAndConquer()) {
+      for (BlobInfo blob : blobs) {
+        DirectDownloadCallable callable = new DirectDownloadCallable(storage, blob, config, opts);
+        downloadTasks.add(convert(executor.submit(callable)));
+      }
+    } else {
+      for (BlobInfo blob : blobs) {
+        BlobInfo validatedBlob = retrieveSizeAndGeneration(storage, blob, config.getBucketName());
+        Path destPath = TransferManagerUtils.createDestPath(config, validatedBlob);
+        long size = validatedBlob.getSize();
+        if (qos.divideAndConquer(size)) {
+
+          DownloadResult optimisticResult =
+              DownloadResult.newBuilder(validatedBlob, TransferStatus.SUCCESS)
+                  .setOutputDestination(destPath)
+                  .build();
+
+          List<ApiFuture<DownloadSegment>> downloadSegmentTasks =
+              computeRanges(size, transferManagerConfig.getPerWorkerBufferSize()).stream()
+                  .map(
+                      r ->
+                          new ChunkedDownloadCallable(
+                              storage, validatedBlob, opts, destPath, r.begin, r.end))
+                  .map(executor::submit)
+                  .map(TransferManagerImpl::convert)
+                  .collect(ImmutableList.toImmutableList());
+
+          downloadTasks.add(
+              ApiFutures.transform(
+                  ApiFutures.allAsList(downloadSegmentTasks),
+                  segments ->
+                      segments.stream()
+                          .reduce(
+                              optimisticResult,
+                              DownloadSegment::reduce,
+                              BinaryOperator.minBy(DownloadResult.COMPARATOR)),
+                  MoreExecutors.directExecutor()));
+        } else {
+          DirectDownloadCallable callable = new DirectDownloadCallable(storage, blob, config, opts);
+          downloadTasks.add(convert(executor.submit(callable)));
+        }
+      }
     }
     return DownloadJob.newBuilder()
         .setDownloadResults(downloadTasks)
@@ -84,4 +135,41 @@ final class TransferManagerImpl implements TransferManager {
   private static <T> ApiFuture<T> convert(ListenableFuture<T> lf) {
     return new ListenableFutureToApiFuture<>(lf);
   }
+
+  private static BlobInfo retrieveSizeAndGeneration(
+      Storage storage, BlobInfo blobInfo, String bucketName) {
+    if (blobInfo.getGeneration() == null) {
+      return storage.get(BlobId.of(bucketName, blobInfo.getName()));
+    } else if (blobInfo.getSize() == null) {
+      return storage.get(BlobId.of(bucketName, blobInfo.getName(), blobInfo.getGeneration()));
+    }
+    return blobInfo;
+  }
+
+  private static ImmutableList<Range> computeRanges(long end, long segmentSize) {
+    ImmutableList.Builder<Range> b = ImmutableList.builder();
+
+    if (end <= segmentSize) {
+      b.add(Range.of(0, end));
+    } else {
+      for (long i = 0; i < end; i += segmentSize) {
+        b.add(Range.of(i, Math.min(i + segmentSize, end)));
+      }
+    }
+    return b.build();
+  }
+
+  private static final class Range {
+    private final long begin;
+    private final long end;
+
+    private Range(long begin, long end) {
+      this.begin = begin;
+      this.end = end;
+    }
+
+    public static Range of(long begin, long end) {
+      return new Range(begin, end);
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerUtils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerUtils.java
index 071e5fc3..607dd314 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerUtils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerUtils.java
@@ -16,6 +16,10 @@
 
 package com.google.cloud.storage.transfermanager;
 
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.StorageException;
+import java.io.IOException;
+import java.nio.file.Files;
 import java.nio.file.Path;
 
 final class TransferManagerUtils {
@@ -29,4 +33,23 @@ final class TransferManagerUtils {
       return config.getPrefix().concat(file.toString());
     }
   }
+
+  static Path createDestPath(ParallelDownloadConfig config, BlobInfo originalBlob) {
+    Path newPath =
+        config
+            .getDownloadDirectory()
+            .resolve(originalBlob.getName().replaceFirst(config.getStripPrefix(), ""));
+    // Check to make sure the parent directories exist
+    if (Files.exists(newPath.getParent())) {
+      return newPath;
+    } else {
+      // Make parent directories if they do not exist
+      try {
+        Files.createDirectories(newPath.getParent());
+        return newPath;
+      } catch (IOException e) {
+        throw new StorageException(e);
+      }
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
index 2bb06330..41ad3e8f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
@@ -16,10 +16,16 @@
 
 package com.google.cloud.storage.transfermanager;
 
+import java.util.Comparator;
+
 public enum TransferStatus {
-  SUCCESS,
   FAILED_TO_START,
   FAILED_TO_FINISH,
   SKIPPED,
-  CANCELED
+  CANCELED,
+  SUCCESS;
+
+  /** A null value is considered to be greater than all values */
+  static final Comparator<TransferStatus> COMPARE_NULL_SAFE =
+      Comparator.nullsLast(Comparator.comparingInt(TransferStatus::ordinal));
 }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
index a16f02da..3d5ab877 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
@@ -26,6 +26,7 @@ import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.TestUtils;
 import com.google.cloud.storage.TmpFile;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
@@ -39,6 +40,7 @@ import com.google.cloud.storage.transfermanager.ParallelDownloadConfig;
 import com.google.cloud.storage.transfermanager.ParallelUploadConfig;
 import com.google.cloud.storage.transfermanager.TransferManager;
 import com.google.cloud.storage.transfermanager.TransferManagerConfig;
+import com.google.cloud.storage.transfermanager.TransferManagerConfigTestingInstances;
 import com.google.cloud.storage.transfermanager.UploadJob;
 import com.google.cloud.storage.transfermanager.UploadResult;
 import com.google.common.collect.ImmutableList;
@@ -48,8 +50,9 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -58,9 +61,15 @@ import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
 @CrossRun(
-    transports = {Transport.HTTP},
+    transports = {Transport.HTTP, Transport.GRPC},
     backends = {Backend.PROD})
 public class ITTransferManagerTest {
+
+  private static final Comparator<BlobInfo> comp =
+      Comparator.comparing(info -> info.getBlobId().getName());
+  private static final Comparator<DownloadResult> comp2 =
+      Comparator.comparing(DownloadResult::getInput, comp);
+
   @Inject public Storage storage;
   @Inject public BucketInfo bucket;
   @Inject public Generator generator;
@@ -82,24 +91,32 @@ public class ITTransferManagerTest {
         BlobInfo.newBuilder(
                 BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
             .build();
-    BlobInfo blobInfo3 =
+    BlobInfo blobInfoChunking =
         BlobInfo.newBuilder(
                 BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
             .build();
-    Collections.addAll(blobs, blobInfo1, blobInfo2, blobInfo3);
+    Collections.addAll(blobs, blobInfo1, blobInfo2);
     ByteBuffer content = DataGenerator.base64Characters().genByteBuffer(108);
     for (BlobInfo blob : blobs) {
       try (WriteChannel writeChannel = storage.writer(blob)) {
         writeChannel.write(content);
       }
     }
+    long size = 2L * 1024 * 1024;
+    size = size + 100L;
+    ByteBuffer chunkedContent = DataGenerator.base64Characters().genByteBuffer(size);
+    try (WriteChannel writeChannel = storage.writer(blobInfoChunking)) {
+      writeChannel.write(chunkedContent);
+    }
+    blobs.add(blobInfoChunking);
   }
 
   @Test
-  public void uploadFiles() throws IOException, ExecutionException, InterruptedException {
-    TransferManagerConfig config = TransferManagerConfig.newBuilder().setMaxWorkers(1).build();
-    TransferManager transferManager = config.getService();
-    try (TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+  public void uploadFiles() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         TmpFile tmpFile1 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         TmpFile tmpFile2 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
       List<Path> files =
@@ -114,10 +131,11 @@ public class ITTransferManagerTest {
   }
 
   @Test
-  public void uploadFilesWithOpts() throws IOException, ExecutionException, InterruptedException {
-    TransferManagerConfig config = TransferManagerConfig.newBuilder().setMaxWorkers(1).build();
-    TransferManager transferManager = config.getService();
-    try (TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+  public void uploadFilesWithOpts() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         TmpFile tmpFile1 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         TmpFile tmpFile2 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
       List<Path> files =
@@ -135,16 +153,67 @@ public class ITTransferManagerTest {
   }
 
   @Test
-  public void downloadBlobs() throws IOException, ExecutionException, InterruptedException {
-    TransferManagerConfig config = TransferManagerConfig.newBuilder().setMaxWorkers(1).build();
-    TransferManager transferManager = config.getService();
-    String bucketName = bucket.getName();
-    ParallelDownloadConfig parallelDownloadConfig =
-        ParallelDownloadConfig.newBuilder().setBucketName(bucketName).build();
-    DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
-    List<DownloadResult> downloadResults = ApiFutures.allAsList(job.getDownloadResults()).get();
-    assertThat(downloadResults).hasSize(3);
-    cleanUpFiles(downloadResults);
+  public void downloadBlobs() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = ApiFutures.allAsList(job.getDownloadResults()).get();
+      try {
+        assertThat(downloadResults).hasSize(3);
+      } finally {
+        cleanUpFiles(downloadResults);
+      }
+    }
+  }
+
+  @Test
+  public void downloadBlobsAllowChunked() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = ApiFutures.allAsList(job.getDownloadResults()).get();
+      assertThat(downloadResults).hasSize(3);
+
+      List<String> expectedContents =
+          blobs.stream()
+              .sorted(comp)
+              .map(BlobInfo::getBlobId)
+              .map(storage::readAllBytes)
+              .map(TestUtils::xxd)
+              .collect(Collectors.toList());
+
+      List<String> actualContents =
+          downloadResults.stream()
+              .sorted(comp2)
+              .map(DownloadResult::getOutputDestination)
+              .map(ITTransferManagerTest::readAllPathBytes)
+              .map(TestUtils::xxd)
+              .collect(Collectors.toList());
+
+      try {
+        assertThat(actualContents).isEqualTo(expectedContents);
+      } finally {
+        cleanUpFiles(downloadResults);
+      }
+    }
   }
 
   private void cleanUpFiles(List<DownloadResult> results) throws IOException {
@@ -154,4 +223,12 @@ public class ITTransferManagerTest {
       Files.delete(res.getOutputDestination().getParent());
     }
   }
+
+  private static byte[] readAllPathBytes(Path path) {
+    try {
+      return Files.readAllBytes(path);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
new file mode 100644
index 00000000..1c5e67b9
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+import com.google.cloud.storage.StorageOptions;
+
+public final class TransferManagerConfigTestingInstances {
+  private TransferManagerConfigTestingInstances() {}
+
+  public static TransferManagerConfig defaults() {
+    return defaults(StorageOptions.newBuilder().build());
+  }
+
+  public static TransferManagerConfig defaults(StorageOptions options) {
+    return TransferManagerConfig.newBuilder()
+        .setAllowDivideAndConquer(false)
+        .setMaxWorkers(1)
+        .setPerWorkerBufferSize(512 * 1024)
+        .setQos(DefaultQos.of(2 * 1024 * 1024))
+        .setStorageOptions(options)
+        .build();
+  }
+}