Repository: googleapis/java-storage
Pull request #2107 (2023-07-07 19:38:18)

fix: update grpc upload logic to follow hashing behavior of json


Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index a4164eb2..14b47899 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -103,7 +103,6 @@ import com.google.storage.v2.NotificationConfig;
 import com.google.storage.v2.NotificationConfigName;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ObjectAccessControl;
-import com.google.storage.v2.ObjectChecksums;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.RewriteObjectRequest;
 import com.google.storage.v2.RewriteResponse;
@@ -233,7 +232,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
+    Hasher hasher = Hasher.enabled();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
@@ -286,8 +285,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
 
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
-
     long size = Files.size(path);
     if (size < bufferSize) {
       // ignore the bufferSize argument if the file is smaller than it
@@ -300,7 +297,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                 ResumableMedia.gapic()
                     .write()
                     .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge))
-                    .setHasher(hasher)
+                    .setHasher(Hasher.enabled())
                     .setByteStringStrategy(ByteStringStrategy.noCopy())
                     .direct()
                     .buffered(Buffers.allocate(size))
@@ -323,7 +320,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
               .write()
               .byteChannel(
                   storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-              .setHasher(hasher)
+              .setHasher(Hasher.noop())
               .setByteStringStrategy(ByteStringStrategy.noCopy())
               .resumable()
               .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
@@ -359,13 +356,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
     ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
 
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
             .byteChannel(
                 storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(hasher)
+            .setHasher(Hasher.noop())
             .setByteStringStrategy(ByteStringStrategy.noCopy())
             .resumable()
             .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
@@ -739,7 +735,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
+    Hasher hasher = Hasher.noop();
     return new GrpcBlobWriteChannel(
         storageClient.writeObjectCallable(),
         getOptions(),
@@ -1967,19 +1963,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         Decoder.identity());
   }
 
-  private static Hasher getHasherForRequest(WriteObjectRequest req, Hasher defaultHasher) {
-    if (!req.hasObjectChecksums()) {
-      return defaultHasher;
-    } else {
-      ObjectChecksums checksums = req.getObjectChecksums();
-      if (!checksums.hasCrc32C() && checksums.getMd5Hash().isEmpty()) {
-        return defaultHasher;
-      } else {
-        return Hasher.noop();
-      }
-    }
-  }
-
   @Nullable
   private Blob internalBlobGet(BlobId blob, Opts<ObjectSourceOpt> unwrap) {
     Opts<ObjectSourceOpt> opts = unwrap.resolveFrom(blob).prepend(defaultOpts);
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..88c013a2
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ChunkedDownloadCallable.java
@@ -0,0 +1,98 @@
+/*
+ * 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.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() {
+    long bytesCopied = -1L;
+    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);
+      bytesCopied = ByteStreams.copy(rc, wc);
+      long bytesExpected = endPosition - startPosition;
+      if (bytesCopied != bytesExpected) {
+        return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+            .setException(
+                new StorageException(
+                    0,
+                    "Unexpected end of stream, read "
+                        + bytesCopied
+                        + " expected "
+                        + bytesExpected
+                        + " from object "
+                        + originalBlob.getBlobId().toGsUtilUriWithGeneration()))
+            .build();
+      }
+    } catch (Exception e) {
+      if (bytesCopied == -1) {
+        return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_START)
+            .setException(e)
+            .build();
+      }
+      return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+    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/DirectDownloadCallable.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java
new file mode 100644
index 00000000..0c7ec86a
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java
@@ -0,0 +1,95 @@
+/*
+ * 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.BlobId;
+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.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
+
+final class DirectDownloadCallable implements Callable<DownloadResult> {
+  private final BlobInfo originalBlob;
+
+  private final ParallelDownloadConfig parallelDownloadConfig;
+  private final Storage storage;
+
+  private final Storage.BlobSourceOption[] opts;
+
+  DirectDownloadCallable(
+      Storage storage,
+      BlobInfo originalBlob,
+      ParallelDownloadConfig parallelDownloadConfig,
+      BlobSourceOption[] opts) {
+    this.originalBlob = originalBlob;
+    this.parallelDownloadConfig = parallelDownloadConfig;
+    this.storage = storage;
+    this.opts = opts;
+  }
+
+  @Override
+  public DownloadResult call() {
+    Path path = TransferManagerUtils.createDestPath(parallelDownloadConfig, originalBlob);
+    long bytesCopied = -1L;
+    try (ReadChannel rc =
+        storage.reader(
+            BlobId.of(parallelDownloadConfig.getBucketName(), originalBlob.getName()), opts)) {
+      FileChannel wc =
+          FileChannel.open(
+              path,
+              StandardOpenOption.WRITE,
+              StandardOpenOption.CREATE,
+              StandardOpenOption.TRUNCATE_EXISTING);
+      bytesCopied = ByteStreams.copy(rc, wc);
+      if (originalBlob.getSize() != null) {
+        if (bytesCopied != originalBlob.getSize()) {
+          return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+              .setException(
+                  new StorageException(
+                      0,
+                      "Unexpected end of stream, read "
+                          + bytesCopied
+                          + " expected "
+                          + originalBlob.getSize()
+                          + " from object "
+                          + originalBlob.getBlobId().toGsUtilUriWithGeneration()))
+              .build();
+        }
+      }
+    } catch (Exception e) {
+      if (bytesCopied == -1) {
+        return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_START)
+            .setException(e)
+            .build();
+      }
+      return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+    DownloadResult result =
+        DownloadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
+            .setOutputDestination(path.toAbsolutePath())
+            .build();
+    return result;
+  }
+}
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
new file mode 100644
index 00000000..96fd933b
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
@@ -0,0 +1,156 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * A parallel download job sent to Transfer Manager.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class DownloadJob {
+
+  @NonNull private final List<ApiFuture<DownloadResult>> downloadResults;
+
+  @NonNull private final ParallelDownloadConfig parallelDownloadConfig;
+
+  private DownloadJob(
+      @NonNull List<ApiFuture<DownloadResult>> downloadResults,
+      @NonNull ParallelDownloadConfig parallelDownloadConfig) {
+    this.downloadResults = downloadResults;
+    this.parallelDownloadConfig = parallelDownloadConfig;
+  }
+
+  /**
+   * The list of {@link DownloadResult DownloadResults} for each download request Transfer Manager
+   * executed for this job. Note calling this method will block the invoking thread until all
+   * download requests are complete.
+   *
+   * @see Builder#setDownloadResults(List)
+   */
+  @BetaApi
+  public @NonNull List<DownloadResult> getDownloadResults() {
+    return ApiExceptions.callAndTranslateApiException(ApiFutures.allAsList(downloadResults));
+  }
+
+  /**
+   * The {@link ParallelDownloadConfig} used for this DownloadJob.
+   *
+   * @see Builder#setParallelDownloadConfig(ParallelDownloadConfig)
+   */
+  @BetaApi
+  public @NonNull ParallelDownloadConfig getParallelDownloadConfig() {
+    return parallelDownloadConfig;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof DownloadJob)) {
+      return false;
+    }
+    DownloadJob that = (DownloadJob) o;
+    return downloadResults.equals(that.downloadResults)
+        && parallelDownloadConfig.equals(that.parallelDownloadConfig);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(downloadResults, parallelDownloadConfig);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("downloadResults", downloadResults)
+        .add("parallelDownloadConfig", parallelDownloadConfig)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /**
+   * Builds an instance of DownloadJob
+   *
+   * @see DownloadJob
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private @NonNull List<ApiFuture<DownloadResult>> downloadResults;
+    private @MonotonicNonNull ParallelDownloadConfig parallelDownloadConfig;
+
+    private Builder() {
+      this.downloadResults = ImmutableList.of();
+    }
+
+    /**
+     * Sets the results for a DownloadJob being performed by Transfer Manager.
+     *
+     * @return the instance of the Builder with DownloadResults modified.
+     * @see DownloadJob#getDownloadResults()
+     */
+    @BetaApi
+    public Builder setDownloadResults(@NonNull List<ApiFuture<DownloadResult>> downloadResults) {
+      this.downloadResults = ImmutableList.copyOf(downloadResults);
+      return this;
+    }
+
+    /**
+     * Sets the {@link ParallelDownloadConfig} used for this DownloadJob.
+     *
+     * @return the instance of the Builder with ParallelDownloadConfig modified.
+     * @see DownloadJob#getParallelDownloadConfig()
+     */
+    @BetaApi
+    public Builder setParallelDownloadConfig(
+        @NonNull ParallelDownloadConfig parallelDownloadConfig) {
+      this.parallelDownloadConfig = parallelDownloadConfig;
+      return this;
+    }
+
+    /**
+     * Creates a DownloadJob object.
+     *
+     * @return {@link DownloadJob}
+     */
+    @BetaApi
+    public DownloadJob build() {
+      checkNotNull(downloadResults);
+      checkNotNull(parallelDownloadConfig);
+      return new DownloadJob(downloadResults, parallelDownloadConfig);
+    }
+  }
+}
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
new file mode 100644
index 00000000..0ac13bba
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
@@ -0,0 +1,231 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.api.core.BetaApi;
+import com.google.cloud.storage.BlobInfo;
+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;
+
+/**
+ * Result for a single download performed by Transfer Manager.
+ *
+ * @see Builder
+ */
+@BetaApi
+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;
+  @NonNull private final TransferStatus status;
+  @MonotonicNonNull private final Exception exception;
+
+  private DownloadResult(
+      @NonNull BlobInfo input,
+      Path outputDestination,
+      @NonNull TransferStatus status,
+      Exception exception) {
+    this.input = input;
+    this.outputDestination = outputDestination;
+    this.status = status;
+    this.exception = exception;
+  }
+
+  /**
+   * The {@link BlobInfo} for the object requested for download.
+   *
+   * @see Builder#setInput(BlobInfo)
+   */
+  @BetaApi
+  public @NonNull BlobInfo getInput() {
+    return input;
+  }
+
+  /**
+   * The destination on the Filesystem the object has been written to. This field will only be
+   * populated if the Transfer was a {@link TransferStatus#SUCCESS SUCCESS}.
+   *
+   * @see Builder#setOutputDestination(Path)
+   */
+  @BetaApi
+  public @NonNull Path getOutputDestination() {
+    checkState(
+        status == TransferStatus.SUCCESS,
+        "getOutputDestination() is only valid when status is SUCCESS but status was %s",
+        status);
+    return outputDestination;
+  }
+
+  /**
+   * The status of the download operation.
+   *
+   * @see TransferStatus
+   * @see Builder#setStatus(TransferStatus)
+   */
+  @BetaApi
+  public @NonNull TransferStatus getStatus() {
+    return status;
+  }
+
+  /**
+   * The exception produced by a failed download operation. This field will only be populated if the
+   * Transfer was not {@link TransferStatus#SUCCESS success}ful or {@link TransferStatus#SKIPPED
+   * skipped}
+   *
+   * @see Builder#setException(Exception)
+   */
+  @BetaApi
+  public @NonNull Exception getException() {
+    checkState(
+        status == TransferStatus.FAILED_TO_FINISH || status == TransferStatus.FAILED_TO_START,
+        "getException() is only valid when an unexpected error has occurred but status was %s",
+        status);
+    return exception;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    DownloadResult that = (DownloadResult) o;
+    return input.equals(that.input)
+        && outputDestination.equals(that.outputDestination)
+        && status == that.status
+        && exception.equals(that.exception);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(input, outputDestination, status, exception);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("input", input)
+        .add("outputDestination", outputDestination)
+        .add("status", status)
+        .add("exception", exception)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder(@NonNull BlobInfo blobInfo, @NonNull TransferStatus status) {
+    return new Builder(blobInfo, status);
+  }
+
+  /**
+   * Builds an instance of DownloadResult
+   *
+   * @see DownloadResult
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private @NonNull BlobInfo input;
+    private @MonotonicNonNull Path outputDestination;
+    private @NonNull TransferStatus status;
+    private @MonotonicNonNull Exception exception;
+
+    private Builder(@NonNull BlobInfo input, @NonNull TransferStatus status) {
+      this.input = input;
+      this.status = status;
+    }
+
+    /**
+     * Sets the {@link BlobInfo} for the object request for download. This field is required.
+     *
+     * @see DownloadResult#getInput()
+     * @return the instance of the Builder with the value for input modified.
+     */
+    @BetaApi
+    public Builder setInput(@NonNull BlobInfo input) {
+      this.input = input;
+      return this;
+    }
+
+    /**
+     * Sets the location on the Filesystem the object has been written to. This field will only be
+     * populated if the Transfer was {@link TransferStatus#SUCCESS success}ful.
+     *
+     * @see DownloadResult#getOutputDestination()
+     * @return the instance of the Builder with the value for outputDestination modified.
+     */
+    @BetaApi
+    public Builder setOutputDestination(@NonNull Path outputDestination) {
+      this.outputDestination = outputDestination;
+      return this;
+    }
+
+    /**
+     * Sets the status of the download.This field is required.
+     *
+     * @see TransferStatus
+     * @return the instance of the Builder with the value for status modified.
+     */
+    @BetaApi
+    public Builder setStatus(@NonNull TransferStatus status) {
+      this.status = status;
+      return this;
+    }
+
+    /**
+     * Sets the Exception produced by a failed download operation. This field will only be populated
+     * if the Transfer was not {@link TransferStatus#SUCCESS success}ful or {@link
+     * TransferStatus#SKIPPED skipped}
+     *
+     * @see DownloadResult#getException()
+     * @return the instance of the Builder with the value for exception modified.
+     */
+    @BetaApi
+    public Builder setException(@NonNull Exception exception) {
+      this.exception = exception;
+      return this;
+    }
+
+    /**
+     * Creates a DownloadResult object.
+     *
+     * @return {@link DownloadResult}
+     */
+    @BetaApi
+    public DownloadResult build() {
+      checkNotNull(input);
+      checkNotNull(status);
+      if (status == TransferStatus.SUCCESS) {
+        checkNotNull(outputDestination);
+      } else if (status == TransferStatus.FAILED_TO_START
+          || status == TransferStatus.FAILED_TO_FINISH) {
+        checkNotNull(exception);
+      }
+      return new DownloadResult(input, outputDestination, status, exception);
+    }
+  }
+}
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..68dcf678
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadSegment.java
@@ -0,0 +1,134 @@
+/*
+ * 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 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 Exception exception;
+
+  private final Long generation;
+
+  private DownloadSegment(
+      BlobInfo input,
+      Path outputDestination,
+      TransferStatus status,
+      Exception 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 Exception 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(result.getStatus(), segment.getStatus()) <= 0) {
+      return result;
+    } else {
+      return segment.toResult();
+    }
+  }
+
+  public static final class Builder {
+
+    private BlobInfo input;
+    private Path outputDestination;
+    private TransferStatus status;
+    private Exception 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(Exception 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/ParallelDownloadConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java
new file mode 100644
index 00000000..222dc9e3
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java
@@ -0,0 +1,216 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.core.BetaApi;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * Configuration for performing Parallel Downloads with {@link TransferManager}.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class ParallelDownloadConfig {
+
+  @NonNull private final String stripPrefix;
+  @NonNull private final Path downloadDirectory;
+  @NonNull private final String bucketName;
+  @NonNull private final List<BlobSourceOption> optionsPerRequest;
+
+  private ParallelDownloadConfig(
+      @NonNull String stripPrefix,
+      @NonNull Path downloadDirectory,
+      @NonNull String bucketName,
+      @NonNull List<BlobSourceOption> optionsPerRequest) {
+    this.stripPrefix = stripPrefix;
+    this.downloadDirectory = downloadDirectory;
+    this.bucketName = bucketName;
+    this.optionsPerRequest = optionsPerRequest;
+  }
+
+  /**
+   * A common prefix removed from an object's name before being written to the filesystem.
+   *
+   * @see Builder#setStripPrefix(String)
+   */
+  @BetaApi
+  public @NonNull String getStripPrefix() {
+    return stripPrefix;
+  }
+
+  /**
+   * The base directory in which all objects will be placed when downloaded.
+   *
+   * @see Builder#setDownloadDirectory(Path)
+   */
+  @BetaApi
+  public @NonNull Path getDownloadDirectory() {
+    return downloadDirectory;
+  }
+
+  /**
+   * The bucket objects are being downloaded from.
+   *
+   * @see Builder#setBucketName(String)
+   */
+  @BetaApi
+  public @NonNull String getBucketName() {
+    return bucketName;
+  }
+
+  /**
+   * A list of common BlobSourceOptions that are used for each download request. Note this list of
+   * options will be applied to every single request.
+   *
+   * @see Builder#setOptionsPerRequest(List)
+   */
+  @BetaApi
+  public @NonNull List<BlobSourceOption> getOptionsPerRequest() {
+    return optionsPerRequest;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof ParallelDownloadConfig)) {
+      return false;
+    }
+    ParallelDownloadConfig that = (ParallelDownloadConfig) o;
+    return stripPrefix.equals(that.stripPrefix)
+        && downloadDirectory.equals(that.downloadDirectory)
+        && bucketName.equals(that.bucketName)
+        && optionsPerRequest.equals(that.optionsPerRequest);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("stripPrefix", stripPrefix)
+        .add("downloadDirectory", downloadDirectory)
+        .add("bucketName", bucketName)
+        .add("optionsPerRequest", optionsPerRequest)
+        .toString();
+  }
+
+  /**
+   * Builds an instance of ParallelDownloadConfig.
+   *
+   * @see ParallelDownloadConfig
+   */
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  @BetaApi
+  public static final class Builder {
+
+    @NonNull private String stripPrefix;
+    @NonNull private Path downloadDirectory;
+    @NonNull private String bucketName;
+    @NonNull private List<BlobSourceOption> optionsPerRequest;
+
+    private Builder() {
+      this.stripPrefix = "";
+      this.downloadDirectory = Paths.get("");
+      this.bucketName = "";
+      this.optionsPerRequest = ImmutableList.of();
+    }
+
+    /**
+     * Sets the value for stripPrefix. This string will be removed from the beginning of all object
+     * names before they are written to the filesystem.
+     *
+     * @return the builder instance with the value for stripPrefix modified.
+     * @see ParallelDownloadConfig#getStripPrefix()
+     */
+    @BetaApi
+    public Builder setStripPrefix(String stripPrefix) {
+      this.stripPrefix = stripPrefix;
+      return this;
+    }
+
+    /**
+     * Sets the base directory on the filesystem that all objects will be written to.
+     *
+     * @return the builder instance with the value for downloadDirectory modified.
+     * @see ParallelDownloadConfig#getDownloadDirectory()
+     */
+    @BetaApi
+    public Builder setDownloadDirectory(Path downloadDirectory) {
+      this.downloadDirectory = downloadDirectory;
+      return this;
+    }
+
+    /**
+     * Sets the bucketName that Transfer Manager will download from. This field is required.
+     *
+     * @return the builder instance with the value for bucketName modified.
+     * @see ParallelDownloadConfig#getBucketName()
+     */
+    @BetaApi
+    public Builder setBucketName(String bucketName) {
+      this.bucketName = bucketName;
+      return this;
+    }
+
+    /**
+     * Sets the BlobSourceOptions that will be applied to each download request. Note these options
+     * will be applied to every single download request.
+     *
+     * @return the builder instance with the value for OptionsPerRequest modified.
+     * @see ParallelDownloadConfig#getOptionsPerRequest()
+     */
+    @BetaApi
+    public Builder setOptionsPerRequest(List<BlobSourceOption> optionsPerRequest) {
+      this.optionsPerRequest = ImmutableList.copyOf(optionsPerRequest);
+      return this;
+    }
+
+    /**
+     * Creates a ParallelDownloadConfig object.
+     *
+     * @return {@link ParallelDownloadConfig}
+     */
+    @BetaApi
+    public ParallelDownloadConfig build() {
+      checkNotNull(bucketName);
+      checkNotNull(stripPrefix);
+      checkNotNull(downloadDirectory);
+      checkNotNull(optionsPerRequest);
+      return new ParallelDownloadConfig(
+          stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
new file mode 100644
index 00000000..b724d981
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
@@ -0,0 +1,221 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.core.BetaApi;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * Configuration for performing Parallel Uploads with {@link TransferManager}.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class ParallelUploadConfig {
+
+  private final boolean skipIfExists;
+  @NonNull private final String prefix;
+  @NonNull private final String bucketName;
+
+  @NonNull private final List<BlobWriteOption> writeOptsPerRequest;
+
+  private ParallelUploadConfig(
+      boolean skipIfExists,
+      @NonNull String prefix,
+      @NonNull String bucketName,
+      @NonNull List<BlobWriteOption> writeOptsPerRequest) {
+    this.skipIfExists = skipIfExists;
+    this.prefix = prefix;
+    this.bucketName = bucketName;
+    this.writeOptsPerRequest = applySkipIfExists(skipIfExists, writeOptsPerRequest);
+  }
+
+  /**
+   * If set Transfer Manager will skip uploading an object if it already exists, equivalent to
+   * providing {@link BlobWriteOption#doesNotExist()} in {@link #getWriteOptsPerRequest()}
+   *
+   * @see Builder#setSkipIfExists(boolean)
+   */
+  @BetaApi
+  public boolean isSkipIfExists() {
+    return skipIfExists;
+  }
+
+  /**
+   * A common prefix that will be applied to all object paths in the destination bucket
+   *
+   * @see Builder#setPrefix(String)
+   */
+  @BetaApi
+  public @NonNull String getPrefix() {
+    return prefix;
+  }
+
+  /**
+   * The bucket objects are being uploaded from
+   *
+   * @see Builder#setBucketName(String)
+   */
+  @BetaApi
+  public @NonNull String getBucketName() {
+    return bucketName;
+  }
+
+  /**
+   * A list of common BlobWriteOptions, note these options will be applied to each upload request.
+   *
+   * @see Builder#setWriteOptsPerRequest(List)
+   */
+  @BetaApi
+  public @NonNull List<BlobWriteOption> getWriteOptsPerRequest() {
+    return writeOptsPerRequest;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ParallelUploadConfig that = (ParallelUploadConfig) o;
+    return skipIfExists == that.skipIfExists
+        && prefix.equals(that.prefix)
+        && bucketName.equals(that.bucketName)
+        && writeOptsPerRequest.equals(that.writeOptsPerRequest);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(skipIfExists, prefix, bucketName, writeOptsPerRequest);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("skipIfExists", skipIfExists)
+        .add("prefix", prefix)
+        .add("bucketName", bucketName)
+        .add("writeOptsPerRequest", writeOptsPerRequest)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  private static List<BlobWriteOption> applySkipIfExists(
+      boolean skipIfExists, List<BlobWriteOption> writeOptsPerRequest) {
+    if (skipIfExists) {
+      return ImmutableList.copyOf(
+          BlobWriteOption.dedupe(writeOptsPerRequest, BlobWriteOption.doesNotExist()));
+    }
+    return writeOptsPerRequest;
+  }
+
+  /**
+   * Builds an instance of ParallelUploadConfig.
+   *
+   * @see ParallelUploadConfig
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private boolean skipIfExists;
+    private @NonNull String prefix;
+    private @NonNull String bucketName;
+    private @NonNull List<BlobWriteOption> writeOptsPerRequest;
+
+    private Builder() {
+      this.prefix = "";
+      this.bucketName = "";
+      this.writeOptsPerRequest = ImmutableList.of();
+    }
+
+    /**
+     * Sets the parameter for skipIfExists. When set to true Transfer Manager will skip uploading an
+     * object if it already exists.
+     *
+     * @return the builder instance with the value for skipIfExists modified.
+     * @see ParallelUploadConfig#isSkipIfExists()
+     */
+    @BetaApi
+    public Builder setSkipIfExists(boolean skipIfExists) {
+      this.skipIfExists = skipIfExists;
+      return this;
+    }
+
+    /**
+     * Sets a common prefix that will be applied to all object paths in the destination bucket.
+     *
+     * @return the builder instance with the value for prefix modified.
+     * @see ParallelUploadConfig#getPrefix()
+     */
+    @BetaApi
+    public Builder setPrefix(@NonNull String prefix) {
+      this.prefix = prefix;
+      return this;
+    }
+
+    /**
+     * Sets the bucketName that Transfer Manager will upload to. This field is required.
+     *
+     * @return the builder instance with the value for bucketName modified.
+     * @see ParallelUploadConfig#getBucketName()
+     */
+    @BetaApi
+    public Builder setBucketName(@NonNull String bucketName) {
+      this.bucketName = bucketName;
+      return this;
+    }
+
+    /**
+     * Sets the BlobWriteOptions that will be applied to each upload request. Note these options
+     * will be applied to every single upload request.
+     *
+     * @return the builder instance with the value for WriteOptsPerRequest modified.
+     * @see ParallelUploadConfig#getWriteOptsPerRequest()
+     */
+    @BetaApi
+    public Builder setWriteOptsPerRequest(@NonNull List<BlobWriteOption> writeOptsPerRequest) {
+      this.writeOptsPerRequest = writeOptsPerRequest;
+      return this;
+    }
+
+    /**
+     * Creates a ParallelUploadConfig object.
+     *
+     * @return {@link ParallelUploadConfig}
+     */
+    @BetaApi
+    public ParallelUploadConfig build() {
+      checkNotNull(prefix);
+      checkNotNull(bucketName);
+      checkNotNull(writeOptsPerRequest);
+      return new ParallelUploadConfig(skipIfExists, prefix, bucketName, writeOptsPerRequest);
+    }
+  }
+}
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
new file mode 100644
index 00000000..f333deca
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
@@ -0,0 +1,94 @@
+/*
+ * 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.api.core.BetaApi;
+import com.google.cloud.storage.BlobInfo;
+import java.nio.file.Path;
+import java.util.List;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * An interface for a Transfer Manager.
+ *
+ * <p>Transfer Manager handles Parallel Uploads and Parallel Downloads.
+ */
+@BetaApi
+public interface TransferManager extends AutoCloseable {
+
+  /**
+   * Uploads a list of files in parallel. This operation will not block the invoking thread,
+   * awaiting results should be done on the returned UploadJob.
+   *
+   * <p>Accepts a {@link ParallelUploadConfig} which defines the constraints of parallel uploads or
+   * predefined defaults.
+   *
+   * <p>Example of creating a parallel upload with Transfer Manager.
+   *
+   * <pre>{@code
+   * String bucketName = "my-unique-bucket";
+   * Path filePath = Paths.get("/path/to/my/file.txt");
+   * Path anotherFilePath = Paths.get("/path/to/another/file.txt");
+   * List<Path> files = List.of(filePath, anotherFilePath);
+   *
+   * ParallelUploadConfig parallelUploadConfig =
+   *           ParallelUploadConfig.newBuilder()
+   *               .setBucketName(bucketName)
+   *               .build();
+   *
+   * UploadJob uploadedFiles = transferManager.uploadFiles(files, config);
+   *
+   * }</pre>
+   *
+   * @return an {@link UploadJob}
+   */
+  @BetaApi
+  @NonNull
+  UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config);
+
+  /**
+   * Downloads a list of blobs in parallel. This operation will not block the invoking thread,
+   * awaiting results should be done on the returned DownloadJob.
+   *
+   * <p>Accepts a {@link ParallelDownloadConfig} which defines the constraints of parallel downloads
+   * or predefined defaults.
+   *
+   * <p>Example of creating a parallel download with Transfer Manager.
+   *
+   * <pre>{@code
+   * String bucketName = "my-unique-bucket";
+   * String blobName = "my-blob-name";
+   * BlobId blobId = BlobId.of(bucketName, blobName);
+   * BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+   * Path baseDir = Paths.get("/path/to/directory/");
+   *
+   * ParallelDownloadConfig parallelDownloadConfig =
+   *           ParallelDownloadConfig.newBuilder()
+   *               .setBucketName(bucketName)
+   *               .setDownloadDirectory(baseDir)
+   *               .build();
+   *
+   * DownloadJob downloadedBlobs = transferManager.downloadBlobs(files, config);
+   *
+   * }</pre>
+   *
+   * @return a {@link DownloadJob}
+   */
+  @BetaApi
+  @NonNull
+  DownloadJob downloadBlobs(List<BlobInfo> blobs, ParallelDownloadConfig config);
+}
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
new file mode 100644
index 00000000..835d5271
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
@@ -0,0 +1,245 @@
+/*
+ * 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.api.core.BetaApi;
+import com.google.cloud.storage.StorageOptions;
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+
+/**
+ * Configuration for an instance of {@link TransferManager}
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class TransferManagerConfig {
+  private final int maxWorkers;
+  private final int perWorkerBufferSize;
+  private final boolean allowDivideAndConquer;
+
+  private final StorageOptions storageOptions;
+  private final Qos qos;
+
+  TransferManagerConfig(
+      int maxWorkers,
+      int perWorkerBufferSize,
+      boolean allowDivideAndConquer,
+      StorageOptions storageOptions,
+      Qos qos) {
+    this.maxWorkers = maxWorkers;
+    this.perWorkerBufferSize = perWorkerBufferSize;
+    this.allowDivideAndConquer = allowDivideAndConquer;
+    this.storageOptions = storageOptions;
+    this.qos = qos;
+  }
+
+  /**
+   * Maximum amount of workers to be allocated to perform work in Transfer Manager
+   *
+   * @see Builder#setMaxWorkers(int)
+   */
+  @BetaApi
+  public int getMaxWorkers() {
+    return maxWorkers;
+  }
+
+  /**
+   * Buffer size allowed to each worker
+   *
+   * @see Builder#setPerWorkerBufferSize(int)
+   */
+  @BetaApi
+  public int getPerWorkerBufferSize() {
+    return perWorkerBufferSize;
+  }
+
+  /**
+   * Whether to allow Transfer Manager to perform chunked Uploads/Downloads if it determines
+   * chunking will be beneficial
+   *
+   * @see Builder#setAllowDivideAndConquer(boolean)
+   */
+  @BetaApi
+  public boolean isAllowDivideAndConquer() {
+    return allowDivideAndConquer;
+  }
+
+  /**
+   * Storage options that Transfer Manager will use to interact with Google Cloud Storage
+   *
+   * @see Builder#setStorageOptions(StorageOptions)
+   */
+  @BetaApi
+  public StorageOptions getStorageOptions() {
+    return storageOptions;
+  }
+
+  /** The service object for {@link TransferManager} */
+  @BetaApi
+  public TransferManager getService() {
+    return new TransferManagerImpl(this);
+  }
+
+  @BetaApi
+  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) {
+      return true;
+    }
+    if (!(o instanceof TransferManagerConfig)) {
+      return false;
+    }
+    TransferManagerConfig that = (TransferManagerConfig) o;
+    return maxWorkers == that.maxWorkers
+        && perWorkerBufferSize == that.perWorkerBufferSize
+        && allowDivideAndConquer == that.allowDivideAndConquer
+        && Objects.equals(storageOptions, that.storageOptions);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(maxWorkers, perWorkerBufferSize, allowDivideAndConquer, storageOptions);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("maxWorkers", maxWorkers)
+        .add("perWorkerBufferSize", perWorkerBufferSize)
+        .add("allowChunking", allowDivideAndConquer)
+        .add("storageOptions", storageOptions)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /**
+   * Builds an instance of TransferManagerConfig
+   *
+   * @see TransferManagerConfig
+   */
+  @BetaApi
+  public static class Builder {
+
+    private int maxWorkers;
+    private int perWorkerBufferSize;
+    private boolean allowDivideAndConquer;
+
+    private StorageOptions storageOptions;
+    private Qos qos;
+
+    private Builder() {
+      this.perWorkerBufferSize = 16 * 1024 * 1024;
+      this.maxWorkers = 2 * Runtime.getRuntime().availableProcessors();
+      this.allowDivideAndConquer = false;
+      this.storageOptions = StorageOptions.getDefaultInstance();
+      this.qos = DefaultQos.of();
+    }
+
+    /**
+     * Maximum amount of workers to be allocated to perform work in Transfer Manager
+     *
+     * <p><i>Default Value:</i> {@code 2 * }{@link Runtime#getRuntime()}{@code .}{@link
+     * Runtime#availableProcessors() availableProcessors()}
+     *
+     * @return the instance of Builder with the value for maxWorkers modified.
+     * @see TransferManagerConfig#getMaxWorkers()
+     */
+    @BetaApi
+    public Builder setMaxWorkers(int maxWorkers) {
+      this.maxWorkers = maxWorkers;
+      return this;
+    }
+
+    /**
+     * Buffer size allowed to each worker
+     *
+     * <p><i>Default Value:</i> 16MiB
+     *
+     * @return the instance of Builder with the value for maxWorkers modified.
+     * @see TransferManagerConfig#getPerWorkerBufferSize()
+     */
+    @BetaApi
+    public Builder setPerWorkerBufferSize(int perWorkerBufferSize) {
+      this.perWorkerBufferSize = perWorkerBufferSize;
+      return this;
+    }
+
+    /**
+     * Whether to allow Transfer Manager to perform chunked Uploads/Downloads if it determines
+     * chunking will be beneficial
+     *
+     * <p><i>Default Value:</i> false
+     *
+     * @return the instance of Builder with the value for allowDivideAndConquer modified.
+     * @see TransferManagerConfig#isAllowDivideAndConquer()
+     */
+    @BetaApi
+    public Builder setAllowDivideAndConquer(boolean allowDivideAndConquer) {
+      this.allowDivideAndConquer = allowDivideAndConquer;
+      return this;
+    }
+
+    /**
+     * Storage options that Transfer Manager will use to interact with Google Cloud Storage
+     *
+     * <p><i>Default Value:</i> {@link StorageOptions#getDefaultInstance()}
+     *
+     * @return the instance of Builder with the value for storageOptions modified.
+     * @see TransferManagerConfig#getStorageOptions()
+     */
+    @BetaApi
+    public Builder setStorageOptions(StorageOptions storageOptions) {
+      this.storageOptions = storageOptions;
+      return this;
+    }
+
+    @BetaApi
+    Builder setQos(Qos qos) {
+      this.qos = qos;
+      return this;
+    }
+
+    /**
+     * Creates a TransferManagerConfig object.
+     *
+     * @return {@link TransferManagerConfig}
+     */
+    @BetaApi
+    public TransferManagerConfig build() {
+      return new TransferManagerConfig(
+          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
new file mode 100644
index 00000000..3d5845cf
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
@@ -0,0 +1,196 @@
+/*
+ * 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.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.core.ListenableFutureToApiFuture;
+import com.google.api.gax.core.GaxProperties;
+import com.google.api.gax.rpc.FixedHeaderProvider;
+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;
+import com.google.cloud.storage.StorageOptions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.nio.file.Files;
+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;
+
+@BetaApi
+final class TransferManagerImpl implements TransferManager {
+
+  private static final String USER_AGENT_ENTRY = "gcloud-tm/";
+  private static final String LIBRARY_VERSION =
+      GaxProperties.getLibraryVersion(TransferManagerConfig.class);
+  private final TransferManagerConfig transferManagerConfig;
+  private final ListeningExecutorService executor;
+  private final Qos qos;
+  private final Storage storage;
+
+  TransferManagerImpl(TransferManagerConfig transferManagerConfig) {
+    this.transferManagerConfig = transferManagerConfig;
+    this.executor =
+        MoreExecutors.listeningDecorator(
+            Executors.newFixedThreadPool(transferManagerConfig.getMaxWorkers()));
+    this.qos = transferManagerConfig.getQos();
+    StorageOptions storageOptions = transferManagerConfig.getStorageOptions();
+    String userAgent = storageOptions.getUserAgent();
+    if (userAgent == null || !userAgent.contains(USER_AGENT_ENTRY)) {
+      storageOptions =
+          storageOptions
+              .toBuilder()
+              .setHeaderProvider(
+                  FixedHeaderProvider.create(
+                      ImmutableMap.of("User-Agent", USER_AGENT_ENTRY + LIBRARY_VERSION)))
+              .build();
+    }
+    this.storage = storageOptions.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
+  @BetaApi
+  public @NonNull UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config) {
+    Storage.BlobWriteOption[] opts =
+        config.getWriteOptsPerRequest().toArray(new BlobWriteOption[0]);
+    List<ApiFuture<UploadResult>> uploadTasks = new ArrayList<>();
+    for (Path file : files) {
+      if (Files.isDirectory(file)) throw new IllegalStateException("Directories are not supported");
+      String blobName = TransferManagerUtils.createBlobName(config, file);
+      BlobInfo blobInfo = BlobInfo.newBuilder(config.getBucketName(), blobName).build();
+      UploadCallable callable =
+          new UploadCallable(transferManagerConfig, storage, blobInfo, file, config, opts);
+      uploadTasks.add(convert(executor.submit(callable)));
+    }
+    return UploadJob.newBuilder()
+        .setParallelUploadConfig(config)
+        .setUploadResults(ImmutableList.copyOf(uploadTasks))
+        .build();
+  }
+
+  @Override
+  @BetaApi
+  public @NonNull DownloadJob downloadBlobs(List<BlobInfo> blobs, ParallelDownloadConfig config) {
+    Storage.BlobSourceOption[] opts =
+        config.getOptionsPerRequest().toArray(new Storage.BlobSourceOption[0]);
+    List<ApiFuture<DownloadResult>> downloadTasks = new ArrayList<>();
+    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, blob);
+        if (validatedBlob != null && qos.divideAndConquer(validatedBlob.getSize())) {
+          DownloadResult optimisticResult =
+              DownloadResult.newBuilder(validatedBlob, TransferStatus.SUCCESS)
+                  .setOutputDestination(destPath)
+                  .build();
+
+          List<ApiFuture<DownloadSegment>> downloadSegmentTasks =
+              computeRanges(validatedBlob.getSize(), 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)
+        .setParallelDownloadConfig(config)
+        .build();
+  }
+
+  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
new file mode 100644
index 00000000..607dd314
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerUtils.java
@@ -0,0 +1,55 @@
+/*
+ * 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.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+final class TransferManagerUtils {
+
+  private TransferManagerUtils() {}
+
+  static String createBlobName(ParallelUploadConfig config, Path file) {
+    if (config.getPrefix().isEmpty()) {
+      return file.toString();
+    } else {
+      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
new file mode 100644
index 00000000..e3b5f4f1
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
@@ -0,0 +1,45 @@
+/*
+ * 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.api.core.BetaApi;
+import java.util.Comparator;
+
+/** The status of a Upload/Download operation performed by Transfer Manager. */
+@BetaApi
+public enum TransferStatus {
+  /** The transfer failed before bytes could be moved. */
+  @BetaApi
+  FAILED_TO_START,
+  /** The transfer failed after bytes could be moved. */
+  @BetaApi
+  FAILED_TO_FINISH,
+  /**
+   * The transfer failed because the object/file already exists and skipIfExists was set to true.
+   *
+   * @see ParallelUploadConfig.Builder#setSkipIfExists(boolean)
+   */
+  @BetaApi
+  SKIPPED,
+  /** The transfer was successful. */
+  @BetaApi
+  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));
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadCallable.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadCallable.java
new file mode 100644
index 00000000..c2b2ff21
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadCallable.java
@@ -0,0 +1,82 @@
+/*
+ * 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.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+import java.nio.file.Path;
+import java.util.concurrent.Callable;
+
+final class UploadCallable implements Callable<UploadResult> {
+  private final TransferManagerConfig transferManagerConfig;
+  private final Storage storage;
+
+  private final BlobInfo originalBlob;
+
+  private final Path sourceFile;
+
+  private final ParallelUploadConfig parallelUploadConfig;
+
+  private final Storage.BlobWriteOption[] opts;
+
+  public UploadCallable(
+      TransferManagerConfig transferManagerConfig,
+      Storage storage,
+      BlobInfo originalBlob,
+      Path sourceFile,
+      ParallelUploadConfig parallelUploadConfig,
+      BlobWriteOption[] opts) {
+    this.transferManagerConfig = transferManagerConfig;
+    this.storage = storage;
+    this.originalBlob = originalBlob;
+    this.sourceFile = sourceFile;
+    this.parallelUploadConfig = parallelUploadConfig;
+    this.opts = opts;
+  }
+
+  public UploadResult call() throws Exception {
+    // TODO: Check for chunking
+    return uploadWithoutChunking();
+  }
+
+  private UploadResult uploadWithoutChunking() {
+    try {
+      Blob from = storage.createFrom(originalBlob, sourceFile, opts);
+      return UploadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
+          .setUploadedBlob(from.asBlobInfo())
+          .build();
+    } catch (StorageException e) {
+      if (parallelUploadConfig.isSkipIfExists() && e.getCode() == 412) {
+        return UploadResult.newBuilder(originalBlob, TransferStatus.SKIPPED)
+            .setException(e)
+            .build();
+      } else {
+        // TODO: check for FAILED_TO_START conditions
+        return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+            .setException(e)
+            .build();
+      }
+    } catch (Exception e) {
+      return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java
new file mode 100644
index 00000000..1b1050b0
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java
@@ -0,0 +1,156 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * A parallel upload job sent to Transfer Manager.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class UploadJob {
+
+  @NonNull private final List<ApiFuture<UploadResult>> uploadResults;
+
+  @NonNull private final ParallelUploadConfig parallelUploadConfig;
+
+  private UploadJob(
+      @NonNull List<ApiFuture<UploadResult>> uploadResults,
+      @NonNull ParallelUploadConfig parallelUploadConfig) {
+    this.uploadResults = uploadResults;
+    this.parallelUploadConfig = parallelUploadConfig;
+  }
+
+  /**
+   * The list of {@link UploadResult UploadResults} for each upload request Transfer Manager
+   * executed for this job. Note calling this method will block the invoking thread until all upload
+   * requests are complete.
+   *
+   * @see Builder#setUploadResults(List)
+   */
+  @BetaApi
+  public List<UploadResult> getUploadResults() {
+    return ApiExceptions.callAndTranslateApiException(ApiFutures.allAsList(uploadResults));
+  }
+
+  /**
+   * The {@link ParallelUploadConfig} used for this UploadJob.
+   *
+   * @see Builder#setParallelUploadConfig(ParallelUploadConfig)
+   */
+  @BetaApi
+  public ParallelUploadConfig getParallelUploadConfig() {
+    return parallelUploadConfig;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof UploadJob)) {
+      return false;
+    }
+    UploadJob uploadJob = (UploadJob) o;
+    return uploadResults.equals(uploadJob.uploadResults)
+        && parallelUploadConfig.equals(uploadJob.parallelUploadConfig);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(uploadResults, parallelUploadConfig);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("uploadResults", uploadResults)
+        .add("parallelUploadConfig", parallelUploadConfig)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /**
+   * Builds an instance of UploadJob
+   *
+   * @see UploadJob
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private @NonNull List<ApiFuture<UploadResult>> uploadResults;
+
+    private @MonotonicNonNull ParallelUploadConfig parallelUploadConfig;
+
+    private Builder() {
+      this.uploadResults = ImmutableList.of();
+    }
+
+    /**
+     * Sets the results for a UploadJob being performed by Transfer Manager.
+     *
+     * @return the instance of the Builder with UploadResults modified.
+     * @see UploadJob#getUploadResults()
+     */
+    @BetaApi
+    public Builder setUploadResults(@NonNull List<ApiFuture<UploadResult>> uploadResults) {
+      this.uploadResults = ImmutableList.copyOf(uploadResults);
+      return this;
+    }
+
+    /**
+     * Sets the {@link ParallelUploadConfig} used for this UploadJob.
+     *
+     * @return the instance of the Builder with ParallelUploadConfig modified.
+     * @see UploadJob#getParallelUploadConfig()
+     */
+    @BetaApi
+    public Builder setParallelUploadConfig(@NonNull ParallelUploadConfig parallelUploadConfig) {
+      this.parallelUploadConfig = parallelUploadConfig;
+      return this;
+    }
+
+    /**
+     * Creates a UploadJob object.
+     *
+     * @return {@link UploadJob}
+     */
+    @BetaApi
+    public UploadJob build() {
+      checkNotNull(uploadResults);
+      checkNotNull(parallelUploadConfig);
+      return new UploadJob(uploadResults, parallelUploadConfig);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java
new file mode 100644
index 00000000..3b9ba38f
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java
@@ -0,0 +1,227 @@
+/*
+ * 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 static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.api.core.BetaApi;
+import com.google.cloud.storage.BlobInfo;
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * Result for a single upload performed by Transfer Manager.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class UploadResult {
+
+  @NonNull private final BlobInfo input;
+  @NonNull private final TransferStatus status;
+  @MonotonicNonNull private final BlobInfo uploadedBlob;
+  @MonotonicNonNull private final Exception exception;
+
+  private UploadResult(
+      @NonNull BlobInfo input,
+      @NonNull TransferStatus status,
+      BlobInfo uploadedBlob,
+      Exception exception) {
+    this.input = input;
+    this.status = status;
+    this.uploadedBlob = uploadedBlob;
+    this.exception = exception;
+  }
+
+  /**
+   * The {@link BlobInfo} for the object requested for upload.
+   *
+   * @see Builder#setInput(BlobInfo)
+   */
+  @BetaApi
+  public @NonNull BlobInfo getInput() {
+    return input;
+  }
+
+  /**
+   * The status of the upload operation.
+   *
+   * @see TransferStatus
+   * @see Builder#setStatus(TransferStatus)
+   */
+  @BetaApi
+  public @NonNull TransferStatus getStatus() {
+    return status;
+  }
+
+  /**
+   * The {@link BlobInfo} for the Uploaded object. This field will only be populated if the Transfer
+   * was {@link TransferStatus#SUCCESS success}ful.
+   *
+   * @see Builder#setUploadedBlob(BlobInfo)
+   */
+  @BetaApi
+  public @NonNull BlobInfo getUploadedBlob() {
+    checkState(
+        status == TransferStatus.SUCCESS,
+        "getUploadedBlob() only valid when status is SUCCESS but status was %s",
+        status);
+    return uploadedBlob;
+  }
+
+  /**
+   * The exception produced by a failed upload operation. This field will only be populated if the
+   * Transfer was not {@link TransferStatus#SUCCESS success}ful or {@link TransferStatus#SKIPPED
+   * skipped}
+   *
+   * @see Builder#setException(Exception)
+   */
+  @BetaApi
+  public @NonNull Exception getException() {
+    checkState(
+        status == TransferStatus.FAILED_TO_START || status == TransferStatus.FAILED_TO_FINISH,
+        "getException() is only valid when an unexpected error has occurred but status was %s",
+        status);
+    return exception;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    UploadResult that = (UploadResult) o;
+    return input.equals(that.input)
+        && status == that.status
+        && uploadedBlob.equals(that.uploadedBlob)
+        && exception.equals(that.exception);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(input, status, uploadedBlob, exception);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("input", input)
+        .add("status", status)
+        .add("uploadedBlob", uploadedBlob)
+        .add("exception", exception)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder(@NonNull BlobInfo input, @NonNull TransferStatus status) {
+    return new Builder(input, status);
+  }
+
+  /**
+   * Builds an instance of UploadResult
+   *
+   * @see UploadResult
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private @NonNull BlobInfo input;
+    private @NonNull TransferStatus status;
+    private @MonotonicNonNull BlobInfo uploadedBlob;
+    private @MonotonicNonNull Exception exception;
+
+    private Builder(@NonNull BlobInfo input, @NonNull TransferStatus status) {
+      this.input = input;
+      this.status = status;
+    }
+
+    /**
+     * Sets the {@link BlobInfo} for the object request for upload. This field is required.
+     *
+     * @return the Builder instance with the value for BlobInfo modified.
+     * @see UploadResult#getInput()
+     */
+    @BetaApi
+    public Builder setInput(@NonNull BlobInfo input) {
+      this.input = input;
+      return this;
+    }
+
+    /**
+     * Sets the Status of the Upload request. This field is required.
+     *
+     * @return the Builder instance with the value for status modified.
+     * @see TransferStatus
+     * @see UploadResult#getStatus()
+     */
+    @BetaApi
+    public Builder setStatus(@NonNull TransferStatus status) {
+      this.status = status;
+      return this;
+    }
+
+    /**
+     * Sets the {@link BlobInfo} for the uploaded object. This field will only be populated if the
+     * Transfer was a {@link TransferStatus#SUCCESS SUCCESS}.
+     *
+     * @return the Builder instance with the value for uploadedBlob modified.
+     */
+    @BetaApi
+    public Builder setUploadedBlob(@NonNull BlobInfo uploadedBlob) {
+      this.uploadedBlob = uploadedBlob;
+      return this;
+    }
+
+    /**
+     * Sets the exception produced by a failed upload operation. This field will only be populated
+     * if the Transfer was not {@link TransferStatus#SUCCESS success}ful or {@link
+     * TransferStatus#SKIPPED skipped}
+     *
+     * @see UploadResult#getException()
+     * @return the Builder instance with the value for exception modified.
+     */
+    @BetaApi
+    public Builder setException(@NonNull Exception exception) {
+      this.exception = exception;
+      return this;
+    }
+
+    /**
+     * Creates an UploadResult object.
+     *
+     * @return {@link UploadResult}
+     */
+    @BetaApi
+    public UploadResult build() {
+      checkNotNull(input);
+      checkNotNull(status);
+      if (status == TransferStatus.SUCCESS) {
+        checkNotNull(uploadedBlob);
+      } else if (status == TransferStatus.FAILED_TO_START
+          || status == TransferStatus.FAILED_TO_FINISH) {
+        checkNotNull(exception);
+      }
+      return new UploadResult(input, status, uploadedBlob, exception);
+    }
+  }
+}
diff --git a/pom.xml b/pom.xml
index d8c1426e..f84766b6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,7 +54,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <github.global.server>github</github.global.server>
     <site.installationModule>google-cloud-storage-parent</site.installationModule>
-    <google.cloud.shared-dependencies.version>3.12.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>3.13.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <dependencyManagement>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
index ad3621e6..ac192a1f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
@@ -241,12 +241,11 @@ public final class ITGrpcStorageImplUploadRetryTest {
         StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
 
     private static final ChecksummedData checksummedData =
-        TestUtils.getChecksummedData(ByteString.copyFrom(bytes), Hasher.enabled());
+        TestUtils.getChecksummedData(ByteString.copyFrom(bytes), Hasher.noop());
     private static final WriteObjectRequest req1 =
         WriteObjectRequest.newBuilder()
             .setUploadId(uploadId)
             .setChecksummedData(checksummedData)
-            .setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(checksummedData.getCrc32C()))
             .setFinishWrite(true)
             .build();
     private static final WriteObjectResponse resp1 =
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
new file mode 100644
index 00000000..7416109b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
@@ -0,0 +1,468 @@
+/*
+ * 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.it;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+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;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.transfermanager.DownloadJob;
+import com.google.cloud.storage.transfermanager.DownloadResult;
+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.TransferStatus;
+import com.google.cloud.storage.transfermanager.UploadJob;
+import com.google.cloud.storage.transfermanager.UploadResult;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    transports = {Transport.HTTP},
+    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;
+
+  @Rule public final TemporaryFolder tmpDir = new TemporaryFolder();
+
+  private Path baseDir;
+  private static final int objectContentSize = 64;
+  private List<BlobInfo> blobs = new ArrayList<>();
+
+  @Before
+  public void setUp() throws Exception {
+    baseDir = tmpDir.getRoot().toPath();
+    BlobInfo blobInfo1 =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    BlobInfo blobInfo2 =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    BlobInfo blobInfoChunking =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    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 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 =
+          ImmutableList.of(tmpFile.getPath(), tmpFile1.getPath(), tmpFile2.getPath());
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(3);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  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 =
+          ImmutableList.of(tmpFile.getPath(), tmpFile1.getPath(), tmpFile2.getPath());
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setWriteOptsPerRequest(Collections.singletonList(BlobWriteOption.doesNotExist()))
+              .build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(3);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  public void uploadFilesOneFailure() 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 =
+          ImmutableList.of(
+              tmpFile.getPath(),
+              tmpFile1.getPath(),
+              tmpFile2.getPath(),
+              Paths.get("this-file-does-not-exist.txt"));
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(4);
+      assertThat(
+              uploadResults.stream()
+                  .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_FINISH)
+                  .collect(Collectors.toList()))
+          .hasSize(1);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  public void uploadNonexistentBucket() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName() + "-does-not-exist";
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      List<Path> files = ImmutableList.of(tmpFile.getPath());
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
+      assertThat(uploadResults.get(0).getException()).isInstanceOf(StorageException.class);
+    }
+  }
+
+  @Test
+  public void uploadNonexistentFile() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService()) {
+      List<Path> files = ImmutableList.of(Paths.get("this-file-does-not-exist.txt"));
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
+      assertThat(uploadResults.get(0).getException()).isInstanceOf(NoSuchFileException.class);
+    }
+  }
+
+  @Test
+  public void uploadFailsSkipIfExists() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).setSkipIfExists(true).build();
+      UploadJob jobInitUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+      UploadJob failedSecondUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
+      assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+    }
+  }
+
+  @Test
+  public void uploadSkipIfExistsGenerationOverride() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setSkipIfExists(true)
+              .setWriteOptsPerRequest(ImmutableList.of(BlobWriteOption.generationMatch(5L)))
+              .build();
+      assertThat(parallelUploadConfig.getWriteOptsPerRequest()).hasSize(1);
+      UploadJob jobInitUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+      UploadJob failedSecondUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
+      assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+    }
+  }
+
+  @Test
+  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 = job.getDownloadResults();
+      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 = job.getDownloadResults();
+      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);
+      }
+    }
+  }
+
+  @Test
+  public void downloadNonexistentBucket() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName() + "-does-not-exist";
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsChunkedFail() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName() + "-does-not-exist";
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      assertThat(downloadResults).hasSize(3);
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsPreconditionFailure() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucket.getName())
+              .setDownloadDirectory(baseDir)
+              .setOptionsPerRequest(ImmutableList.of(BlobSourceOption.generationMatch(-1)))
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      assertThat(downloadResults).hasSize(3);
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsOneFailure() 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();
+      List<BlobInfo> downloadBlobs = blobs;
+      BlobInfo nonexistentBlob =
+          BlobInfo.newBuilder(
+                  BlobId.of(
+                      bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+              .build();
+      downloadBlobs.add(nonexistentBlob);
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      try {
+        assertThat(downloadResults).hasSize(4);
+        assertThat(
+                downloadResults.stream()
+                    .filter(res -> res.getStatus() == TransferStatus.FAILED_TO_START)
+                    .collect(Collectors.toList()))
+            .hasSize(1);
+      } finally {
+        cleanUpFiles(
+            downloadResults.stream()
+                .filter(res -> res.getStatus() == TransferStatus.SUCCESS)
+                .collect(Collectors.toList()));
+      }
+    }
+  }
+
+  private void cleanUpFiles(List<DownloadResult> results) throws IOException {
+    // Cleanup downloaded blobs and the parent directory
+    for (DownloadResult res : results) {
+      Files.delete(res.getOutputDestination());
+      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();
+  }
+}