Repository: googleapis/java-storage
Pull request #2105 (2023-07-06 15:25:49)

feat: Initial Preview of Transfer Manager
# Google Cloud Storage Transfer Manager Preview

Transfer Manager is currently in Preview, to denote this we have added the `@BetaApi` annotation to all related Transfer Manager APIs. Using Transfer Manager APIs at this time comes with the possibility of breaking changes in the future. The `@BetaApi` annotation will be removed to denote we are out of preview and the API is stable and will not break outside of a major release.

## Notable features

1. Parallel Uploads for a list of Files.
2. Parallel Downloads for a list of Blobs.
3. Parallel Downloads for large chunkable blobs. (i.e. Divide and Conquer Downloads)

## Not yet Implemented

1. Parallel Uploads for large chunkable files. (i.e. Divide and Conquer Uploads)







Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index 258389d1..73568a1e 100644
--- a/.github/.OwlBot.lock.yaml
+++ b/.github/.OwlBot.lock.yaml
@@ -13,5 +13,5 @@
 # limitations under the License.
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-java:latest
-  digest: sha256:31c8276a1bfb43766597d32645721c029cb94571f1b8d996cb2c290744fe52f9
-# created: 2023-06-27T18:51:58.922150125Z
+  digest: sha256:af2eda87a54601ae7b7b2be5055c17b43ac98a7805b586772db314de8a7d4a1d
+# created: 2023-06-16T02:10:09.149325782Z
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index 203f9eac..fde1ced4 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -6,8 +6,7 @@ updates:
       interval: "daily"
     # Disable version updates for Maven dependencies
     # we use renovate-bot as well as shared-dependencies BOM to update maven dependencies.
-    ignore:
-      - dependency-name: "*"
+    ignore: "*"
   - package-ecosystem: "pip"
     directory: "/"
     schedule:
@@ -15,5 +14,4 @@ updates:
     # Disable version updates for pip dependencies
     # If a security vulnerability comes in, we will be notified about 
     # it via template in the synthtool repository.
-    ignore:
-      - dependency-name: "*"
+    ignore: "*"
diff --git a/.github/release-please.yml b/.github/release-please.yml
index 1280ad03..d71d8f4e 100644
--- a/.github/release-please.yml
+++ b/.github/release-please.yml
@@ -1,8 +1,7 @@
 bumpMinorPreMajor: true
 handleGHRelease: true
 releaseType: java-yoshi
-extraFiles:
-  - README.md
+extraFiles: ["README.md"]
 branches:
   - bumpMinorPreMajor: true
     handleGHRelease: true
@@ -32,9 +31,3 @@ branches:
     handleGHRelease: true
     releaseType: java-backport
     branch: 2.15.x
-  - bumpMinorPreMajor: true
-    handleGHRelease: true
-    releaseType: java-backport
-    extraFiles:
-      - README.md
-    branch: 2.22.x
diff --git a/.github/sync-repo-settings.yaml b/.github/sync-repo-settings.yaml
index 898a8c57..b7d5dcb3 100644
--- a/.github/sync-repo-settings.yaml
+++ b/.github/sync-repo-settings.yaml
@@ -8,7 +8,8 @@ branchProtectionRules:
     requiresCodeOwnerReviews: true
     requiresStrictStatusChecks: false
     requiredStatusCheckContexts:
-      - dependencies (17)
+      - dependencies (8)
+      - dependencies (11)
       - lint
       - clirr
       - units (8)
@@ -128,23 +129,6 @@ branchProtectionRules:
       - cla/google
       - 'Kokoro - Test: Java GraalVM Native Image'
       - 'Kokoro - Test: Java 17 GraalVM Native Image'
-  - pattern: 2.22.x
-    isAdminEnforced: true
-    requiredApprovingReviewCount: 1
-    requiresCodeOwnerReviews: true
-    requiresStrictStatusChecks: false
-    requiredStatusCheckContexts:
-      - dependencies (8)
-      - dependencies (11)
-      - lint
-      - clirr
-      - units (8)
-      - units (11)
-      - 'Kokoro - Test: Integration'
-      - cla/google
-      - OwlBot Post Processor
-      - 'Kokoro - Test: Java GraalVM Native Image'
-      - 'Kokoro - Test: Java 17 GraalVM Native Image'
 permissionRules:
   - team: yoshi-admins
     permission: admin
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index e27b2c57..e3bb26e3 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -25,39 +25,17 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [11, 17]
+        java: [8, 11, 17]
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
       with:
-        distribution: temurin
+        distribution: zulu
         java-version: ${{matrix.java}}
     - run: java -version
     - run: .kokoro/build.sh
       env:
         JOB_TYPE: test
-  units-java8:
-    # Building using Java 17 and run the tests with Java 8 runtime
-    name: "units (8)"
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-java@v3
-        with:
-          java-version: 8
-          distribution: temurin
-      - name: "Set jvm system property environment variable for surefire plugin (unit tests)"
-        # Maven surefire plugin (unit tests) allows us to specify JVM to run the tests.
-        # https://maven.apache.org/surefire/maven-surefire-plugin/test-mojo.html#jvm
-        run: echo "SUREFIRE_JVM_OPT=-Djvm=${JAVA_HOME}/bin/java" >> $GITHUB_ENV
-        shell: bash
-      - uses: actions/setup-java@v3
-        with:
-          java-version: 17
-          distribution: temurin
-      - run: .kokoro/build.sh
-        env:
-          JOB_TYPE: test
   windows:
     runs-on: windows-latest
     steps:
@@ -66,7 +44,7 @@ jobs:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
       with:
-        distribution: temurin
+        distribution: zulu
         java-version: 8
     - run: java -version
     - run: .kokoro/build.bat
@@ -76,34 +54,22 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        java: [17]
+        java: [8, 11, 17]
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
       with:
-        distribution: temurin
+        distribution: zulu
         java-version: ${{matrix.java}}
     - run: java -version
     - run: .kokoro/dependencies.sh
-  javadoc:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v3
-    - uses: actions/setup-java@v3
-      with:
-        distribution: temurin
-        java-version: 17
-    - run: java -version
-    - run: .kokoro/build.sh
-      env:
-        JOB_TYPE: javadoc
   lint:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
       with:
-        distribution: temurin
+        distribution: zulu
         java-version: 11
     - run: java -version
     - run: .kokoro/build.sh
@@ -115,7 +81,7 @@ jobs:
     - uses: actions/checkout@v3
     - uses: actions/setup-java@v3
       with:
-        distribution: temurin
+        distribution: zulu
         java-version: 8
     - run: java -version
     - run: .kokoro/build.sh
diff --git a/.github/workflows/samples.yaml b/.github/workflows/samples.yaml
index 10d252d7..912ed8b2 100644
--- a/.github/workflows/samples.yaml
+++ b/.github/workflows/samples.yaml
@@ -23,7 +23,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-java@v3
         with:
-          distribution: temurin
+          distribution: zulu
           java-version: 8
       - name: Run checkstyle
         run: mvn -P lint --quiet --batch-mode checkstyle:check
diff --git a/.kokoro/build.sh b/.kokoro/build.sh
index 49069482..f105d675 100755
--- a/.kokoro/build.sh
+++ b/.kokoro/build.sh
@@ -47,8 +47,7 @@ set +e
 
 case ${JOB_TYPE} in
 test)
-    echo "SUREFIRE_JVM_OPT: ${SUREFIRE_JVM_OPT}"
-    mvn test -B -ntp -Dclirr.skip=true -Denforcer.skip=true ${SUREFIRE_JVM_OPT}
+    mvn test -B -ntp -Dclirr.skip=true -Denforcer.skip=true
     RETURN_CODE=$?
     ;;
 lint)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b4cf5918..cbe9b317 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,19 +1,5 @@
 # Changelog
 
-## [2.23.0](https://github.com/googleapis/java-storage/compare/v2.22.4...v2.23.0) (2023-06-21)
-
-
-### Features
-
-* Add new dedup utility method to Option classes ([#2063](https://github.com/googleapis/java-storage/issues/2063)) ([2ad196c](https://github.com/googleapis/java-storage/commit/2ad196c063e67c7efdac344792b67de3479d789d))
-
-
-### Dependencies
-
-* Update dependencies io.grpc:* to v1.56.0 ([#2072](https://github.com/googleapis/java-storage/issues/2072)) ([e10bde2](https://github.com/googleapis/java-storage/commit/e10bde26416bcf17401516e43949e12246f4831c))
-* Update dependency com.google.apis:google-api-services-storage to v1-rev20230617-2.0.0 ([#2077](https://github.com/googleapis/java-storage/issues/2077)) ([9f618cd](https://github.com/googleapis/java-storage/commit/9f618cddbeb471f7bd0f2332c70e501afbaccc36))
-* Update dependency org.graalvm.buildtools:native-maven-plugin to v0.9.23 ([#2074](https://github.com/googleapis/java-storage/issues/2074)) ([427f330](https://github.com/googleapis/java-storage/commit/427f330793a20b0c3da4cbe5e85984a0df508c79))
-
 ## [2.22.4](https://github.com/googleapis/java-storage/compare/v2.22.3...v2.22.4) (2023-06-07)
 
 
diff --git a/README.md b/README.md
index e40d26ae..1b1d586b 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.18.0</version>
+      <version>26.17.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -42,7 +42,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.23.0</version>
+  <version>2.22.4</version>
 </dependency>
 
 ```
@@ -50,20 +50,20 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.18.0')
+implementation platform('com.google.cloud:libraries-bom:26.17.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.23.0'
+implementation 'com.google.cloud:google-cloud-storage:2.22.4'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.23.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.22.4"
 ```
 <!-- {x-version-update-end} -->
 
@@ -224,8 +224,6 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-storage/tree/
 | Configure Retries | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/ConfigureRetries.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/ConfigureRetries.java) |
 | Generate Signed Post Policy V4 | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/GenerateSignedPostPolicyV4.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/GenerateSignedPostPolicyV4.java) |
 | Get Service Account | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/GetServiceAccount.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/GetServiceAccount.java) |
-| Quickstart Grpc Dp Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java) |
-| Quickstart Grpc Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java) |
 | Quickstart Sample | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/QuickstartSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/QuickstartSample.java) |
 | Add Bucket Default Owner | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/AddBucketDefaultOwner.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/AddBucketDefaultOwner.java) |
 | Add Bucket Iam Conditional Binding | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/AddBucketIamConditionalBinding.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/AddBucketIamConditionalBinding.java) |
@@ -428,7 +426,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.23.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.22.4
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles
diff --git a/gapic-google-cloud-storage-v2/pom.xml b/gapic-google-cloud-storage-v2/pom.xml
index 863c8ff7..e8aa83e0 100644
--- a/gapic-google-cloud-storage-v2/pom.xml
+++ b/gapic-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>gapic-google-cloud-storage-v2</artifactId>
-  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
   <name>gapic-google-cloud-storage-v2</name>
   <description>GRPC library for gapic-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/google-cloud-storage-bom/pom.xml b/google-cloud-storage-bom/pom.xml
index 19635d94..356eca8e 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -19,7 +19,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-bom</artifactId>
-  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>pom</packaging>
   <parent>
     <groupId>com.google.cloud</groupId>
@@ -69,22 +69,22 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
     </dependencies>
   </dependencyManagement>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index af60edb4..aa16af16 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -2,7 +2,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>jar</packaging>
   <name>Google Cloud Storage</name>
   <url>https://github.com/googleapis/java-storage</url>
@@ -12,11 +12,11 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <pubsub-proto.version>1.105.17</pubsub-proto.version>
+    <pubsub-proto.version>1.105.14</pubsub-proto.version>
     <junit-platform.version>5.9.3</junit-platform.version>
   </properties>
   <dependencies>
@@ -173,13 +173,13 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.114.0</version>
+      <version>0.113.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-kms</artifactId>
-      <version>2.23.0</version>
+      <version>2.22.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -248,7 +248,7 @@
     <dependency>
       <groupId>net.jqwik</groupId>
       <artifactId>jqwik</artifactId>
-      <version>1.7.4</version>
+      <version>1.7.3</version>
       <scope>test</scope>
     </dependency>
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
index bb7780c0..d328fa9f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
@@ -1078,10 +1078,10 @@ public class BucketInfo implements Serializable {
       }
 
       /**
-       * Creates a new {@code LifecycleAction}, with no specific supported action associated with
-       * it. This is only intended as a "backup" for when the library doesn't recognize the type,
-       * and should generally not be used, instead use the supported actions, and upgrade the
-       * library if necessary to get new supported actions.
+       * Creates a new {@code LifecycleAction , with no specific supported action associated with it. This
+       * is only intended as a "backup" for when the library doesn't recognize the type, and should
+       * generally not be used, instead use the supported actions, and upgrade the library if necessary
+       * to get new supported actions.
        */
       public static LifecycleAction newLifecycleAction(@NonNull String actionType) {
         return new LifecycleAction(actionType);
@@ -2437,7 +2437,7 @@ public class BucketInfo implements Serializable {
    * Returns the bucket's recovery point objective (RPO). This defines how quickly data is
    * replicated between regions in a dual-region bucket. Not defined for single-region buckets.
    *
-   * @see <a href="https://cloud.google.com/storage/docs/turbo-replication">Turbo Replication"</a>
+   * @see <a href="https://cloud.google.com/storage/docs/turbo-replication"Turbo Replication"</a>
    */
   public Rpo getRpo() {
     return rpo;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
index ca4cd64d..96929b30 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
@@ -29,7 +29,7 @@ import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
- * Typeful sealed class hierarchy for representing an <a target="_blank"
+ * Typeful sealed class hierarchy for representing an <a target="_blank" rel="noopener noreferrer"
  * href="https://developer.mozilla.org/en-US/docs/Web/HTTP/Range_requests#requesting_a_specific_range_from_a_server">HTTP
  * Range Header</a> There are certain subtleties when building these header values depending on
  * whether explicit boundaries or relative lengths are used. This class encapsulates the edge cases
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cUtility.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cUtility.java
index 1a3882da..13269387 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cUtility.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cUtility.java
@@ -30,7 +30,7 @@ class Crc32cUtility {
    * @param crc2 crc32c int of the second object
    * @param crc2ObjectSize byte length of object which generated crc2
    * @return
-   * @see <a target="_blank"
+   * @see <a target="_blank" rel="noopener noreferrer"
    *     href="https://github.com/google/crc32c/blob/21fc8ef30415a635e7351ffa0e5d5367943d4a94/src/crc32c_portable.cc#L252-L257">https://github.com/google/crc32c/blob/21fc8ef30415a635e7351ffa0e5d5367943d4a94/src/crc32c_portable.cc#L252-L257</a>
    */
   public static int crc32cCombineGoogle(int crc1, int crc2, long crc2ObjectSize) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
index 45543f00..3222bc4e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
@@ -25,7 +25,7 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
   /**
    * Concatenate {@code other} to {@code this} value.
    *
-   * <p>The concat operation satisfies the Left <a target="_blank"
+   * <p>The concat operation satisfies the Left <a target="_blank" rel="noopener noreferrer"
    * href="https://en.wikipedia.org/wiki/Distributive_property">Distributive property</a>.
    *
    * <p>This means, given the following instances:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index e39bf85d..a8725609 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -156,7 +156,7 @@ final class GapicWritableByteChannelSessionBuilder {
      * Buffer using {@code byteBuffer} worth of space before attempting to flush.
      *
      * <p>The provided {@link ByteBuffer} <i>should</i> be aligned with GCSs block size of <a
-     * target="_blank"
+     * target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload">256
      * KiB</a> if mu
      */
@@ -252,7 +252,7 @@ final class GapicWritableByteChannelSessionBuilder {
      * Buffer using {@code byteBuffer} worth of space before attempting to flush.
      *
      * <p>The provided {@link ByteBuffer} <i>should</i> be aligned with GCSs block size of <a
-     * target="_blank"
+     * target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload">256
      * KiB</a>.
      */
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index d623745a..1b69e673 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -371,6 +371,9 @@ public final class GrpcStorageOptions extends StorageOptions
     /**
      * Option which signifies the client should attempt to connect to gcs via Direct Path.
      *
+     * <p>In order to use direct path, both this option must be true and the environment variable
+     * (not system property) {@code GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS} must be true.
+     *
      * <p><i>NOTE</i>There is no need to specify a new endpoint via {@link #setHost(String)} as the
      * underlying code will translate the normal {@code https://storage.googleapis.com:443} into the
      * proper Direct Path URI for you.
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
index 967577ed..972a7df6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GzipReadableByteChannel.java
@@ -130,7 +130,7 @@ final class GzipReadableByteChannel implements UnbufferedReadableByteChannel {
    * <p>This class decorates an InputStream to be optimistic about the number of available bytes
    * when reading data to encourage GzipInputStream to consume the entire stream.
    *
-   * <p>For a more in-depth write up see <a target="_blank"
+   * <p>For a more in-depth write up see <a target="_blank" rel="noopener noreferrer"
    * href="https://github.com/googleapis/google-http-java-client/pull/1608">google-http-java-client/pull/1608</a>
    *
    * <p><i>NOTE</i> This class is a copy of the private class from {@code
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index daab166f..63a0793c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -3917,7 +3917,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * <h4>Behavioral Differences between HTTP and gRPC</h4>
    *
    * <ol>
-   *   <li>Calling this method for a Bucket which has <a target="_blank"
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
    *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
    *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
    *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
@@ -4060,7 +4060,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * <h4>Behavioral Differences between HTTP and gRPC</h4>
    *
    * <ol>
-   *   <li>Calling this method for a Bucket which has <a target="_blank"
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
    *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
    *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
    *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
@@ -4095,7 +4095,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * <h4>Behavioral Differences between HTTP and gRPC</h4>
    *
    * <ol>
-   *   <li>Calling this method for a Bucket which has <a target="_blank"
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
    *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
    *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
    *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
@@ -4188,7 +4188,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * <h4>Behavioral Differences between HTTP and gRPC</h4>
    *
    * <ol>
-   *   <li>Calling this method for a Bucket which has <a target="_blank"
+   *   <li>Calling this method for a Bucket which has <a target="_blank" rel="noopener noreferrer"
    *       href="https://cloud.google.com/storage/docs/uniform-bucket-level-access">Uniform
    *       bucket-level access</a> enabled exhibits different behavior Depending on which {@link
    *       Transport} is used. For JSON, an HTTP 400 Bad Request error will be thrown. Whereas for
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageRetryStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageRetryStrategy.java
index d0a340a1..19a332d4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageRetryStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageRetryStrategy.java
@@ -51,8 +51,9 @@ public interface StorageRetryStrategy extends Serializable {
    * <p>The set of retryable cases handled by this strategy is more comprehensive than that of the
    * legacy strategy and should always be preferred.
    *
+   * <p>The following HTTP Status Codes will be retried for all idempotent calls:
+   *
    * <table>
-   *   <caption>Retried HTTP status codes for idempotent calls</caption>
    *   <tr>
    *     <th>Code</th>
    *     <th>Name</th>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java
index 63a95b40..c58cafe7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java
@@ -43,7 +43,7 @@ public @interface TransportCompatibility {
    */
   enum Transport {
     /**
-     * Value indicating use of the <a target="_blank"
+     * Value indicating use of the <a target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/json_api">Cloud Storage JSON API</a>
      *
      * @see StorageOptions#http()
@@ -51,7 +51,7 @@ public @interface TransportCompatibility {
     HTTP,
 
     /**
-     * Value indicating usa of the <a target="_blank"
+     * Value indicating usa of the <a target="_blank" rel="noopener noreferrer"
      * href="https://github.com/googleapis/googleapis/blob/master/google/storage/v2/storage.proto">Cloud
      * Storage v2 gRPC API</a> TODO: link to public docs when published.
      *
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index f51bc05d..7aca3921 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -79,9 +79,10 @@ import org.checkerframework.checker.nullness.qual.RequiresNonNull;
  * The set of all "Options" we currently support for per-call parameters.
  *
  * <p>Most often, each of the respective types correspond to one of the parameters from <a
- * target="_blank" href="https://cloud.google.com/storage/docs/json_api/v1/parameters">HTTP headers
- * and common query string parameters for JSON</a>. In the case of gRPC, sometimes the parameters
- * are in the specific request message or in grpc metadata.
+ * target="_blank" rel="noopener noreferrer"
+ * href="https://cloud.google.com/storage/docs/json_api/v1/parameters">HTTP headers and common query
+ * string parameters for JSON</a>. In the case of gRPC, sometimes the parameters are in the specific
+ * request message or in grpc metadata.
  */
 @SuppressWarnings({"deprecation", "DeprecatedIsStillUsed"})
 final class UnifiedOpts {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index cf4151e2..d5a205ed 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -44,7 +44,8 @@ import org.checkerframework.checker.nullness.qual.Nullable;
  *
  * <p>In this class "fsync" is used to mean "complete the client stream to GCS and await its
  * response". We are loosely following the concept used in linux to flush data to disk <a
- * target="_blank" href="https://man7.org/linux/man-pages/man2/fdatasync.2.html">fsync(2)</a>
+ * target="_blank" rel="noopener noreferrer"
+ * href="https://man7.org/linux/man-pages/man2/fdatasync.2.html">fsync(2)</a>
  */
 final class WriteFlushStrategy {
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index 100f5b6c..18e95e19 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -1103,7 +1103,6 @@ public class HttpStorageRpc implements StorageRpc {
 
       Long maxBytesRewrittenPerCall =
           req.megabytesRewrittenPerCall != null ? req.megabytesRewrittenPerCall * MEGABYTE : null;
-      StorageObject content = req.overrideInfo ? req.target : null;
       Storage.Objects.Rewrite rewrite =
           storage
               .objects()
@@ -1112,7 +1111,7 @@ public class HttpStorageRpc implements StorageRpc {
                   req.source.getName(),
                   req.target.getBucket(),
                   req.target.getName(),
-                  content)
+                  req.overrideInfo ? req.target : null)
               .setSourceGeneration(req.source.getGeneration())
               .setRewriteToken(token)
               .setMaxBytesRewrittenPerCall(maxBytesRewrittenPerCall)
@@ -1133,7 +1132,6 @@ public class HttpStorageRpc implements StorageRpc {
               .setDestinationPredefinedAcl(Option.PREDEFINED_ACL.getString(req.targetOptions))
               .setUserProject(userProject)
               .setDestinationKmsKeyName(Option.KMS_KEY_NAME.getString(req.targetOptions));
-      rewrite.setDisableGZipContent(content == null);
       HttpHeaders requestHeaders = rewrite.getRequestHeaders();
       setEncryptionHeaders(requestHeaders, SOURCE_ENCRYPTION_KEY_PREFIX, req.sourceOptions);
       setEncryptionHeaders(requestHeaders, ENCRYPTION_KEY_PREFIX, req.targetOptions);
@@ -1462,7 +1460,6 @@ public class HttpStorageRpc implements StorageRpc {
           .hmacKeys()
           .create(projectId, serviceAccountEmail)
           .setUserProject(Option.USER_PROJECT.getString(options))
-          .setDisableGZipContent(true)
           .execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
@@ -1717,7 +1714,6 @@ public class HttpStorageRpc implements StorageRpc {
           .buckets()
           .lockRetentionPolicy(bucket.getName(), Option.IF_METAGENERATION_MATCH.getLong(options))
           .setUserProject(Option.USER_PROJECT.getString(options))
-          .setDisableGZipContent(true)
           .execute();
     } catch (IOException ex) {
       span.setStatus(Status.UNKNOWN.withDescription(ex.getMessage()));
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/grpc-google-cloud-storage-v2/pom.xml b/grpc-google-cloud-storage-v2/pom.xml
index 3844d4b1..1a6e1417 100644
--- a/grpc-google-cloud-storage-v2/pom.xml
+++ b/grpc-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-v2</artifactId>
-  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
   <name>grpc-google-cloud-storage-v2</name>
   <description>GRPC library for grpc-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/pom.xml b/pom.xml
index f84766b6..b7444791 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-parent</artifactId>
   <packaging>pom</packaging>
-  <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -54,46 +54,33 @@
     <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.13.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>3.11.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <dependencyManagement>
     <dependencies>
-      <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-bom</artifactId>
-        <version>1.56.1</version>
-        <type>pom</type>
-        <scope>import</scope>
-      </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>
         <version>${google.cloud.shared-dependencies.version}</version>
         <type>pom</type>
         <scope>import</scope>
-        <exclusions>
-          <exclusion>
-            <groupId>io.grpc</groupId>
-            <artifactId>*</artifactId>
-          </exclusion>
-        </exclusions>
       </dependency>
 
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.apis</groupId>
         <artifactId>google-api-services-storage</artifactId>
-        <version>v1-rev20230617-2.0.0</version>
+        <version>v1-rev20230301-2.0.0</version>
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-pubsub</artifactId>
-        <version>1.123.17</version>
+        <version>1.123.14</version>
         <scope>test</scope>
       </dependency>
       <dependency>
@@ -124,17 +111,17 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
@@ -177,7 +164,7 @@
       <dependency>
         <groupId>com.google.truth</groupId>
         <artifactId>truth</artifactId>
-        <version>1.1.5</version>
+        <version>1.1.4</version>
         <scope>test</scope>
         <exclusions>
           <exclusion>
diff --git a/proto-google-cloud-storage-v2/pom.xml b/proto-google-cloud-storage-v2/pom.xml
index 1c6a5e2d..dc5d1647 100644
--- a/proto-google-cloud-storage-v2/pom.xml
+++ b/proto-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-v2</artifactId>
-  <version>2.23.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+  <version>2.22.5-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
   <name>proto-google-cloud-storage-v2</name>
   <description>PROTO library for proto-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 770c6866..6ad8b2bb 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,7 +30,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.23.0</version>
+      <version>2.22.4</version>
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
@@ -43,7 +43,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.5</version>
+      <version>1.1.4</version>
       <scope>test</scope>
     </dependency>
       <dependency>
@@ -61,7 +61,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.17</version>
+      <version>1.123.14</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index 841e4a18..14152b6c 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -29,7 +29,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.18.0</version>
+        <version>26.17.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -55,13 +55,13 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.5</version>
+      <version>1.1.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.17</version>
+      <version>1.123.14</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 0127f645..df4f7850 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,7 +28,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.23.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.22.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
 
     <dependency>
@@ -40,7 +40,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.5</version>
+      <version>1.1.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -52,7 +52,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.17</version>
+      <version>1.123.14</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 3f47056d..95f581a7 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.18.0</version>
+        <version>26.17.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -54,7 +54,7 @@
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>
-      <version>1.1.5</version>
+      <version>1.1.4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -72,7 +72,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.123.17</version>
+      <version>1.123.14</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java
deleted file mode 100644
index a657300c..00000000
--- a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcDpSample.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Copyright 2023 Google Inc.
- *
- * 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.example.storage;
-
-// [START storage_grpc_quickstart_dp]
-// Imports the Google Cloud client library
-import com.google.cloud.storage.Bucket;
-import com.google.cloud.storage.BucketInfo;
-import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.StorageOptions;
-
-public class QuickstartGrpcDpSample {
-  public static void main(String... args) throws Exception {
-
-    // Create an instance of options which will use the Google Cloud Storage gRPC API for all
-    // operations
-    StorageOptions options = StorageOptions.grpc().setAttemptDirectPath(true).build();
-
-    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
-    try (Storage storage = options.getService()) {
-      // The name for the new bucket
-      String bucketName = args[0]; // "my-new-bucket";
-
-      // Creates the new bucket using a request to the gRPC API via DirectPath
-      Bucket bucket = storage.create(BucketInfo.of(bucketName));
-
-      System.out.printf("Bucket %s created.%n", bucket.getName());
-    }
-  }
-}
-// [END storage_grpc_quickstart_dp]
diff --git a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java b/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java
deleted file mode 100644
index 84c737ba..00000000
--- a/samples/snippets/src/main/java/com/example/storage/QuickstartGrpcSample.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Copyright 2023 Google Inc.
- *
- * 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.example.storage;
-
-// [START storage_grpc_quickstart]
-// Imports the Google Cloud client library
-import com.google.cloud.storage.Bucket;
-import com.google.cloud.storage.BucketInfo;
-import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.StorageOptions;
-
-public class QuickstartGrpcSample {
-  public static void main(String... args) throws Exception {
-
-    // Create an instance of options which will use the Google Cloud Storage gRPC API for all
-    // operations
-    StorageOptions options = StorageOptions.grpc().build();
-
-    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
-    try (Storage storage = options.getService()) {
-      // The name for the new bucket
-      String bucketName = args[0]; // "my-new-bucket";
-
-      // Creates the new bucket using a request to the gRPC API
-      Bucket bucket = storage.create(BucketInfo.of(bucketName));
-
-      System.out.printf("Bucket %s created.%n", bucket.getName());
-    }
-  }
-}
-// [END storage_grpc_quickstart]
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/DisableDefaultEventBasedHold.java b/samples/snippets/src/main/java/com/example/storage/bucket/DisableDefaultEventBasedHold.java
index a5529d60..744e7fb7 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/DisableDefaultEventBasedHold.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/DisableDefaultEventBasedHold.java
@@ -18,9 +18,8 @@ package com.example.storage.bucket;
 
 // [START storage_disable_default_event_based_hold]
 
-import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
@@ -34,11 +33,7 @@ public class DisableDefaultEventBasedHold {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-    // first look up the bucket, so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
-    storage.update(
-        bucket.toBuilder().setDefaultEventBasedHold(false).build(),
-        BucketTargetOption.metagenerationMatch());
+    storage.update(BucketInfo.newBuilder(bucketName).setDefaultEventBasedHold(false).build());
 
     System.out.println("Default event-based hold was disabled for " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/DisableLifecycleManagement.java b/samples/snippets/src/main/java/com/example/storage/bucket/DisableLifecycleManagement.java
index 6463358e..b0d5f19c 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/DisableLifecycleManagement.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/DisableLifecycleManagement.java
@@ -19,7 +19,6 @@ package com.example.storage.bucket;
 // [START storage_disable_bucket_lifecycle_management]
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageOptions;
 
 public class DisableLifecycleManagement {
@@ -31,11 +30,8 @@ public class DisableLifecycleManagement {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-    // first look up the bucket, so we will have its metageneration
     Bucket bucket = storage.get(bucketName);
-    storage.update(
-        bucket.toBuilder().deleteLifecycleRules().build(),
-        BucketTargetOption.metagenerationMatch());
+    bucket.toBuilder().deleteLifecycleRules().build().update();
 
     System.out.println("Lifecycle management was disabled for bucket " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/DisableUniformBucketLevelAccess.java b/samples/snippets/src/main/java/com/example/storage/bucket/DisableUniformBucketLevelAccess.java
index dc77bc99..8beff201 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/DisableUniformBucketLevelAccess.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/DisableUniformBucketLevelAccess.java
@@ -18,10 +18,8 @@ package com.example.storage.bucket;
 
 // [START storage_disable_uniform_bucket_level_access]
 
-import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
@@ -35,18 +33,12 @@ public class DisableUniformBucketLevelAccess {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-
-    // first look up the bucket, so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
-
     BucketInfo.IamConfiguration iamConfiguration =
         BucketInfo.IamConfiguration.newBuilder()
             .setIsUniformBucketLevelAccessEnabled(false)
             .build();
 
-    storage.update(
-        bucket.toBuilder().setIamConfiguration(iamConfiguration).build(),
-        BucketTargetOption.metagenerationMatch());
+    storage.update(BucketInfo.newBuilder(bucketName).setIamConfiguration(iamConfiguration).build());
 
     System.out.println("Uniform bucket-level access was disabled for " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/EnableDefaultEventBasedHold.java b/samples/snippets/src/main/java/com/example/storage/bucket/EnableDefaultEventBasedHold.java
index 0ae5ba8a..d443f9fc 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/EnableDefaultEventBasedHold.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/EnableDefaultEventBasedHold.java
@@ -18,9 +18,8 @@ package com.example.storage.bucket;
 
 // [START storage_enable_default_event_based_hold]
 
-import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
@@ -34,11 +33,7 @@ public class EnableDefaultEventBasedHold {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-    // first look up the bucket, so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
-    storage.update(
-        bucket.toBuilder().setDefaultEventBasedHold(true).build(),
-        BucketTargetOption.metagenerationMatch());
+    storage.update(BucketInfo.newBuilder(bucketName).setDefaultEventBasedHold(true).build());
 
     System.out.println("Default event-based hold was enabled for " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/EnableUniformBucketLevelAccess.java b/samples/snippets/src/main/java/com/example/storage/bucket/EnableUniformBucketLevelAccess.java
index 4c70b7b2..70975629 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/EnableUniformBucketLevelAccess.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/EnableUniformBucketLevelAccess.java
@@ -18,10 +18,8 @@ package com.example.storage.bucket;
 
 // [START storage_enable_uniform_bucket_level_access]
 
-import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
@@ -35,21 +33,10 @@ public class EnableUniformBucketLevelAccess {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-
-    // first look up the bucket, so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
-
     BucketInfo.IamConfiguration iamConfiguration =
         BucketInfo.IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build();
 
-    storage.update(
-        bucket
-            .toBuilder()
-            .setIamConfiguration(iamConfiguration)
-            .setAcl(null)
-            .setDefaultAcl(null)
-            .build(),
-        BucketTargetOption.metagenerationMatch());
+    storage.update(BucketInfo.newBuilder(bucketName).setIamConfiguration(iamConfiguration).build());
 
     System.out.println("Uniform bucket-level access was enabled for " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/RemoveBucketDefaultKmsKey.java b/samples/snippets/src/main/java/com/example/storage/bucket/RemoveBucketDefaultKmsKey.java
index 26c07475..83ab00d6 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/RemoveBucketDefaultKmsKey.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/RemoveBucketDefaultKmsKey.java
@@ -19,7 +19,6 @@ package com.example.storage.bucket;
 // [START storage_bucket_delete_default_kms_key]
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageOptions;
 
 public class RemoveBucketDefaultKmsKey {
@@ -31,11 +30,8 @@ public class RemoveBucketDefaultKmsKey {
     // String bucketName = "your-unique-bucket-name";
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-    // first look up the bucket, so we will have its metageneration
     Bucket bucket = storage.get(bucketName);
-    storage.update(
-        bucket.toBuilder().setDefaultKmsKeyName(null).build(),
-        BucketTargetOption.metagenerationMatch());
+    bucket.toBuilder().setDefaultKmsKeyName(null).build().update();
 
     System.out.println("Default KMS key was removed from " + bucketName);
   }
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/SetBucketDefaultKmsKey.java b/samples/snippets/src/main/java/com/example/storage/bucket/SetBucketDefaultKmsKey.java
index 995001db..947eb760 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/SetBucketDefaultKmsKey.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/SetBucketDefaultKmsKey.java
@@ -19,8 +19,8 @@ package com.example.storage.bucket;
 // [START storage_set_bucket_default_kms_key]
 
 import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
@@ -38,19 +38,13 @@ public class SetBucketDefaultKmsKey {
     // "projects/your-project-id/locations/us/keyRings/my_key_ring/cryptoKeys/my_key"
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
-    // first look up the bucket, so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
 
-    Bucket updated =
-        storage.update(
-            bucket.toBuilder().setDefaultKmsKeyName(kmsKeyName).build(),
-            BucketTargetOption.metagenerationMatch());
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName).setDefaultKmsKeyName(kmsKeyName).build();
+    Bucket bucket = storage.update(bucketInfo);
 
     System.out.println(
-        "KMS Key "
-            + updated.getDefaultKmsKeyName()
-            + "was set to default for bucket "
-            + bucketName);
+        "KMS Key " + bucket.getDefaultKmsKeyName() + "was set to default for bucket " + bucketName);
   }
 }
 // [END storage_set_bucket_default_kms_key]
diff --git a/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java b/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java
index eca89fbb..1d40f46a 100644
--- a/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java
+++ b/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java
@@ -19,11 +19,10 @@ package com.example.storage.bucket;
 // [START storage_set_retention_policy]
 
 import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
-import java.time.Duration;
 
 public class SetRetentionPolicy {
   public static void setRetentionPolicy(
@@ -39,21 +38,15 @@ public class SetRetentionPolicy {
 
     Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
 
-    // first look up the bucket so we will have its metageneration
-    Bucket bucket = storage.get(bucketName);
     Bucket bucketWithRetentionPolicy =
         storage.update(
-            bucket
-                .toBuilder()
-                .setRetentionPeriodDuration(Duration.ofSeconds(retentionPeriodSeconds))
-                .build(),
-            BucketTargetOption.metagenerationMatch());
+            BucketInfo.newBuilder(bucketName).setRetentionPeriod(retentionPeriodSeconds).build());
 
     System.out.println(
         "Retention period for "
             + bucketName
             + " is now "
-            + bucketWithRetentionPolicy.getRetentionPeriodDuration());
+            + bucketWithRetentionPolicy.getRetentionPeriod());
   }
 }
 // [END storage_set_retention_policy]
diff --git a/versions.txt b/versions.txt
index 819d5aaf..84a23716 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,7 +1,7 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:2.23.0:2.23.1-SNAPSHOT
-gapic-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT
-grpc-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT
-proto-google-cloud-storage-v2:2.23.0-alpha:2.23.1-alpha-SNAPSHOT
+google-cloud-storage:2.22.4:2.22.5-SNAPSHOT
+gapic-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT
+grpc-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT
+proto-google-cloud-storage-v2:2.22.4-alpha:2.22.5-alpha-SNAPSHOT


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index 24ac6a0b..79c6cf35 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -58,7 +58,7 @@ import org.junit.runner.RunWith;
 @RunWith(StorageITRunner.class)
 @CrossRun(
     backends = {Backend.TEST_BENCH},
-    transports = {Transport.GRPC})
+    transports = {Transport.HTTP, Transport.GRPC})
 public final class ITGrpcTest {
 
   @Inject public Storage storage;
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/it/runner/SneakyException.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
index aa6d0059..52790c3d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/SneakyException.java
@@ -23,7 +23,7 @@ package com.google.cloud.storage.it.runner;
  * <p>This class provides some utility methods to sneakily (not method declared) throw exceptions
  * and later unwrap any sneakily wrapped exception if it's needed.
  */
-public final class SneakyException extends RuntimeException {
+final class SneakyException extends RuntimeException {
 
   public SneakyException(Throwable cause) {
     super(cause);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
index bb6d99b4..41c4f497 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
@@ -30,13 +30,9 @@ import com.google.api.core.NanoClock;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.RetryHelper.RetryHelperException;
-import com.google.cloud.Tuple;
 import com.google.cloud.conformance.storage.v1.InstructionList;
 import com.google.cloud.conformance.storage.v1.Method;
-import com.google.cloud.storage.it.runner.SneakyException;
-import com.google.common.base.Charsets;
 import com.google.common.collect.ImmutableList;
-import com.google.common.io.CharStreams;
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonElement;
@@ -45,24 +41,19 @@ import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.net.SocketException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 import org.threeten.bp.Duration;
 
 /**
- * A {@link ManagedLifecycle} which integrates with the <a target="_blank"
+ * A {@link ManagedLifecycle} which integrates with the <a target="_blank" rel="noopener noreferrer"
  * href="https://github.com/googleapis/storage-testbench">storage-testbench</a> by pulling the
  * docker image, starting the container, providing methods for interacting with the {@code
  * /retry_test} rest api, stopping the container.
@@ -420,47 +411,9 @@ public final class TestBench implements ManagedLifecycle {
   static final class Builder {
     private static final String DEFAULT_BASE_URI = "http://localhost:9000";
     private static final String DEFAULT_GRPC_BASE_URI = "http://localhost:9005";
-    private static final String DEFAULT_IMAGE_NAME;
-    private static final String DEFAULT_IMAGE_TAG;
-
-    static {
-      ClassLoader cl = Thread.currentThread().getContextClassLoader();
-      Tuple<String, String> nameAndTag =
-          SneakyException.unwrap(
-                  () -> {
-                    InputStream dockerfileText =
-                        cl.getResourceAsStream(
-                            "com/google/cloud/storage/it/runner/registry/Dockerfile");
-                    //noinspection UnstableApiUsage
-                    return Optional.ofNullable(dockerfileText)
-                        .map(is -> new InputStreamReader(is, Charsets.UTF_8))
-                        .flatMap(
-                            reader ->
-                                SneakyException.sneaky(
-                                    () ->
-                                        CharStreams.readLines(reader).stream()
-                                            .filter(line -> !line.startsWith("#"))
-                                            .filter(line -> line.startsWith("FROM"))
-                                            .findFirst()
-                                            .flatMap(
-                                                from -> {
-                                                  Pattern pattern =
-                                                      Pattern.compile("FROM (.*?):(.*)$");
-                                                  Matcher matcher = pattern.matcher(from);
-                                                  if (matcher.matches()) {
-                                                    return Optional.of(
-                                                        Tuple.of(
-                                                            matcher.group(1), matcher.group(2)));
-                                                  } else {
-                                                    return Optional.empty();
-                                                  }
-                                                })));
-                  })
-              .orElse(Tuple.of(null, null));
-      DEFAULT_IMAGE_NAME = nameAndTag.x();
-      DEFAULT_IMAGE_TAG = nameAndTag.y();
-    }
-
+    private static final String DEFAULT_IMAGE_NAME =
+        "gcr.io/cloud-devrel-public-resources/storage-testbench";
+    private static final String DEFAULT_IMAGE_TAG = "v0.35.0";
     private static final String DEFAULT_CONTAINER_NAME = "default";
 
     private boolean ignorePullError;
@@ -530,8 +483,8 @@ public final class TestBench implements ManagedLifecycle {
           ignorePullError,
           baseUri,
           gRPCBaseUri,
-          requireNonNull(dockerImageName, "dockerImageName must be non null"),
-          requireNonNull(dockerImageTag, "dockerImageTag must be non null"),
+          dockerImageName,
+          dockerImageTag,
           String.format("storage-testbench_%s", containerName));
     }
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/StorageArbitraries.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/StorageArbitraries.java
index 6d93b27e..95aa7284 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/StorageArbitraries.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/jqwik/StorageArbitraries.java
@@ -173,7 +173,8 @@ public final class StorageArbitraries {
     private Buckets() {}
 
     /**
-     * Generated bucket name based on the rules outlined in <a target="_blank"
+     * Generated bucket name based on the rules outlined in <a target="_blank" rel="noopener
+     * noreferrer"
      * href="https://cloud.google.com/storage/docs/naming-buckets#requirements">https://cloud.google.com/storage/docs/naming-buckets#requirements</a>
      */
     public Arbitrary<BucketName> name() {
@@ -453,7 +454,8 @@ public final class StorageArbitraries {
     private Objects() {}
 
     /**
-     * Generated object name based on the rules outlined in <a target="_blank"
+     * Generated object name based on the rules outlined in <a target="_blank" rel="noopener
+     * noreferrer"
      * href="https://cloud.google.com/storage/docs/naming-objects#objectnames">https://cloud.google.com/storage/docs/naming-objects#objectnames</a>
      */
     public Arbitrary<String> name() {
@@ -503,7 +505,7 @@ public final class StorageArbitraries {
     }
 
     /**
-     * Custom metadata from <a target="_blank"
+     * Custom metadata from <a target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/metadata">https://cloud.google.com/storage/docs/metadata</a>
      */
     public Arbitrary<Map<String, String>> customMetadata() {
@@ -527,7 +529,7 @@ public final class StorageArbitraries {
   }
 
   /**
-   * Fixed-key metadata from <a target="_blank"
+   * Fixed-key metadata from <a target="_blank" rel="noopener noreferrer"
    * href="https://cloud.google.com/storage/docs/metadata">https://cloud.google.com/storage/docs/metadata</a>
    */
   public static final class HttpHeaders {
@@ -595,7 +597,7 @@ public final class StorageArbitraries {
     }
 
     /**
-     * <a target="_blank"
+     * <a target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/json_api/v1/objectAccessControls#resource">https://cloud.google.com/storage/docs/json_api/v1/objectAccessControls#resource</a>
      */
     public Arbitrary<String> role() {
@@ -603,7 +605,7 @@ public final class StorageArbitraries {
     }
 
     /**
-     * <a target="_blank"
+     * <a target="_blank" rel="noopener noreferrer"
      * href="https://cloud.google.com/storage/docs/json_api/v1/objectAccessControls#resource">https://cloud.google.com/storage/docs/json_api/v1/objectAccessControls#resource</a>
      */
     public Arbitrary<String> team() {
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();
+  }
+}
diff --git a/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile b/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
deleted file mode 100644
index 35df4e1d..00000000
--- a/google-cloud-storage/src/test/resources/com/google/cloud/storage/it/runner/registry/Dockerfile
+++ /dev/null
@@ -1,1 +0,0 @@
-FROM gcr.io/cloud-devrel-public-resources/storage-testbench:v0.36.0
diff --git a/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java b/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
index 5bc17937..6d434e30 100644
--- a/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
+++ b/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
@@ -16,7 +16,6 @@
 
 package com.example.storage;
 
-import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -72,18 +71,14 @@ import com.example.storage.object.ReleaseEventBasedHold;
 import com.example.storage.object.ReleaseTemporaryHold;
 import com.example.storage.object.SetEventBasedHold;
 import com.example.storage.object.SetTemporaryHold;
-import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.Identity;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.storage.Blob;
-import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketInfo;
-import com.google.cloud.storage.BucketInfo.PublicAccessPrevention;
 import com.google.cloud.storage.Cors;
 import com.google.cloud.storage.HttpMethod;
 import com.google.cloud.storage.Storage;
-import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.StorageRoles;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.cloud.testing.junit4.StdOutCaptureRule;
@@ -98,12 +93,12 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.rules.Timeout;
-import org.threeten.bp.Duration;
 
 public class ITBucketSnippets {
 
@@ -113,13 +108,6 @@ public class ITBucketSnippets {
   private static final String KMS_KEY_NAME =
       "projects/java-docs-samples-testing/locations/us/keyRings/"
           + "jds_test_kms_key_ring/cryptoKeys/gcs_kms_key_one";
-  private static final RetrySettings RETRY_SETTINGS =
-      RetrySettings.newBuilder()
-          .setInitialRetryDelay(Duration.ofSeconds(2))
-          .setRetryDelayMultiplier(1.75)
-          .setTotalTimeout(Duration.ofSeconds(90))
-          .setMaxRetryDelay(Duration.ofSeconds(10))
-          .build();
 
   private static Storage storage;
 
@@ -172,14 +160,12 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testChangeDefaultStorageClass() throws Throwable {
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> assertEquals("STANDARD", storage.get(BUCKET).getStorageClass().name()));
+  public void testChangeDefaultStorageClass() {
+    Bucket remoteBucket = storage.get(BUCKET);
+    assertEquals("STANDARD", remoteBucket.getStorageClass().name());
     ChangeDefaultStorageClass.changeDefaultStorageClass(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> assertEquals("COLDLINE", storage.get(BUCKET).getStorageClass().name()));
+    remoteBucket = storage.get(BUCKET);
+    assertEquals("COLDLINE", remoteBucket.getStorageClass().name());
   }
 
   @Test
@@ -286,14 +272,13 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testEnableLifecycleManagement() throws Throwable {
+  public void testEnableLifecycleManagement() {
     EnableLifecycleManagement.enableLifecycleManagement(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertEquals(1, storage.get(BUCKET).getLifecycleRules().size()));
+    assertEquals(1, storage.get(BUCKET).getLifecycleRules().size());
   }
 
   @Test
-  public void testDisableLifecycleManagement() throws Throwable {
+  public void testDisableLifecycleManagement() {
     storage
         .get(BUCKET)
         .toBuilder()
@@ -304,15 +289,13 @@ public class ITBucketSnippets {
                     BucketInfo.LifecycleRule.LifecycleCondition.newBuilder().setAge(5).build())))
         .build()
         .update();
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertEquals(1, storage.get(BUCKET).getLifecycleRules().size()));
+    assertEquals(1, storage.get(BUCKET).getLifecycleRules().size());
     DisableLifecycleManagement.disableLifecycleManagement(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertEquals(0, storage.get(BUCKET).getLifecycleRules().size()));
+    assertEquals(0, storage.get(BUCKET).getLifecycleRules().size());
   }
 
   @Test
-  public void testGetPublicAccessPrevention() throws Throwable {
+  public void testGetPublicAccessPrevention() {
     try {
       // By default a bucket PAP state is INHERITED and we are changing the state to validate
       // non-default state.
@@ -325,14 +308,18 @@ public class ITBucketSnippets {
                   .build())
           .build()
           .update();
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () ->
-              assertThat(storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention())
-                  .isEqualTo(PublicAccessPrevention.ENFORCED));
       GetPublicAccessPrevention.getPublicAccessPrevention(PROJECT_ID, BUCKET);
       String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
       assertTrue(snippetOutput.contains("enforced"));
+      storage
+          .get(BUCKET)
+          .toBuilder()
+          .setIamConfiguration(
+              BucketInfo.IamConfiguration.newBuilder()
+                  .setPublicAccessPrevention(BucketInfo.PublicAccessPrevention.INHERITED)
+                  .build())
+          .build()
+          .update();
     } finally {
       // No matter what happens make sure test set bucket back to INHERITED
       storage
@@ -348,15 +335,21 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testSetPublicAccessPreventionEnforced() throws Throwable {
+  public void testSetPublicAccessPreventionEnforced() {
     try {
       SetPublicAccessPreventionEnforced.setPublicAccessPreventionEnforced(PROJECT_ID, BUCKET);
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () ->
-              assertEquals(
-                  storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention(),
-                  BucketInfo.PublicAccessPrevention.ENFORCED));
+      assertEquals(
+          storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention(),
+          BucketInfo.PublicAccessPrevention.ENFORCED);
+      storage
+          .get(BUCKET)
+          .toBuilder()
+          .setIamConfiguration(
+              BucketInfo.IamConfiguration.newBuilder()
+                  .setPublicAccessPrevention(BucketInfo.PublicAccessPrevention.INHERITED)
+                  .build())
+          .build()
+          .update();
     } finally {
       // No matter what happens make sure test set bucket back to INHERITED
       storage
@@ -372,7 +365,7 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testSetPublicAccessPreventionInherited() throws Throwable {
+  public void testSetPublicAccessPreventionInherited() {
     try {
       storage
           .get(BUCKET)
@@ -383,19 +376,10 @@ public class ITBucketSnippets {
                   .build())
           .build()
           .update();
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () ->
-              assertThat(storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention())
-                  .isEqualTo(PublicAccessPrevention.ENFORCED));
-
       SetPublicAccessPreventionInherited.setPublicAccessPreventionInherited(PROJECT_ID, BUCKET);
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () ->
-              assertEquals(
-                  storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention(),
-                  BucketInfo.PublicAccessPrevention.INHERITED));
+      assertEquals(
+          storage.get(BUCKET).getIamConfiguration().getPublicAccessPrevention(),
+          BucketInfo.PublicAccessPrevention.INHERITED);
     } finally {
       // No matter what happens make sure test set bucket back to INHERITED
       storage
@@ -411,7 +395,7 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testAddListRemoveBucketIamMembers() throws Throwable {
+  public void testAddListRemoveBucketIamMembers() {
     storage.update(
         BucketInfo.newBuilder(BUCKET)
             .setIamConfiguration(
@@ -419,7 +403,6 @@ public class ITBucketSnippets {
                     .setIsUniformBucketLevelAccessEnabled(true)
                     .build())
             .build());
-    // todo:
     int originalSize = storage.getIamPolicy(BUCKET).getBindingsList().size();
     AddBucketIamMember.addBucketIamMember(PROJECT_ID, BUCKET);
     assertEquals(originalSize + 1, storage.getIamPolicy(BUCKET).getBindingsList().size());
@@ -431,9 +414,7 @@ public class ITBucketSnippets {
     AddBucketIamConditionalBinding.addBucketIamConditionalBinding(PROJECT_ID, BUCKET);
     assertEquals(originalSize + 1, storage.getIamPolicy(BUCKET).getBindingsList().size());
     RemoveBucketIamConditionalBinding.removeBucketIamConditionalBinding(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> assertEquals(originalSize, storage.getIamPolicy(BUCKET).getBindingsList().size()));
+    assertEquals(originalSize, storage.getIamPolicy(BUCKET).getBindingsList().size());
     storage.update(
         BucketInfo.newBuilder(BUCKET)
             .setIamConfiguration(
@@ -444,21 +425,18 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testMakeBucketPublic() throws Throwable {
+  public void testMakeBucketPublic() {
     MakeBucketPublic.makeBucketPublic(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () ->
-            assertTrue(
-                storage
-                    .getIamPolicy(BUCKET)
-                    .getBindings()
-                    .get(StorageRoles.objectViewer())
-                    .contains(Identity.allUsers())));
+    assertTrue(
+        storage
+            .getIamPolicy(BUCKET)
+            .getBindings()
+            .get(StorageRoles.objectViewer())
+            .contains(Identity.allUsers()));
   }
 
   @Test
-  public void deleteBucketDefaultKmsKey() throws Throwable {
+  public void deleteBucketDefaultKmsKey() {
     storage
         .get(BUCKET)
         .toBuilder()
@@ -467,33 +445,25 @@ public class ITBucketSnippets {
                 + "jds_test_kms_key_ring/cryptoKeys/gcs_kms_key_one")
         .build()
         .update();
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertNotNull(storage.get(BUCKET).getDefaultKmsKeyName()));
+    assertNotNull(storage.get(BUCKET).getDefaultKmsKeyName());
     RemoveBucketDefaultKmsKey.removeBucketDefaultKmsKey(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertNull(storage.get(BUCKET).getDefaultKmsKeyName()));
+    assertNull(storage.get(BUCKET).getDefaultKmsKeyName());
   }
 
   @Test
-  public void testEnableDisableVersioning() throws Throwable {
+  public void testEnableDisableVersioning() {
     EnableBucketVersioning.enableBucketVersioning(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertTrue(storage.get(BUCKET).versioningEnabled()));
+    assertTrue(storage.get(BUCKET).versioningEnabled());
     DisableBucketVersioning.disableBucketVersioning(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertFalse(storage.get(BUCKET).versioningEnabled()));
+    Assert.assertFalse(storage.get(BUCKET).versioningEnabled());
   }
 
   @Test
-  public void testSetBucketWebsiteInfo() throws Throwable {
+  public void testSetBucketWebsiteInfo() {
     SetBucketWebsiteInfo.setBucketWesbiteInfo(PROJECT_ID, BUCKET, "index.html", "404.html");
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> {
-          Bucket bucket = storage.get(BUCKET);
-          assertEquals("index.html", bucket.getIndexPage());
-          assertEquals("404.html", bucket.getNotFoundPage());
-        });
+    Bucket bucket = storage.get(BUCKET);
+    assertEquals("index.html", bucket.getIndexPage());
+    assertEquals("404.html", bucket.getNotFoundPage());
   }
 
   @Test
@@ -504,22 +474,18 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testConfigureBucketCors() throws Throwable {
+  public void testConfigureBucketCors() {
     ConfigureBucketCors.configureBucketCors(
         PROJECT_ID, BUCKET, "http://example.appspot.com", "Content-Type", 3600);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> {
-          Cors cors = storage.get(BUCKET).getCors().get(0);
-          assertTrue(cors.getOrigins().get(0).toString().contains("example.appspot.com"));
-          assertTrue(cors.getResponseHeaders().contains("Content-Type"));
-          assertEquals(3600, cors.getMaxAgeSeconds().intValue());
-          assertTrue(cors.getMethods().get(0).toString().equalsIgnoreCase("GET"));
-        });
+    Cors cors = storage.get(BUCKET).getCors().get(0);
+    assertTrue(cors.getOrigins().get(0).toString().contains("example.appspot.com"));
+    assertTrue(cors.getResponseHeaders().contains("Content-Type"));
+    assertEquals(3600, cors.getMaxAgeSeconds().intValue());
+    assertTrue(cors.getMethods().get(0).toString().equalsIgnoreCase("GET"));
   }
 
   @Test
-  public void testRemoveBucketCors() throws Throwable {
+  public void testRemoveBucketCors() {
     storage
         .get(BUCKET)
         .toBuilder()
@@ -533,66 +499,51 @@ public class ITBucketSnippets {
                     .build()))
         .build()
         .update();
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> {
-          Cors cors = storage.get(BUCKET).getCors().get(0);
-          assertNotNull(cors);
-          assertTrue(cors.getOrigins().get(0).toString().contains("example.appspot.com"));
-          assertTrue(cors.getResponseHeaders().contains("Content-Type"));
-          assertEquals(3600, cors.getMaxAgeSeconds().intValue());
-          assertTrue(cors.getMethods().get(0).toString().equalsIgnoreCase("GET"));
-        });
+    Cors cors = storage.get(BUCKET).getCors().get(0);
+    assertNotNull(cors);
+    assertTrue(cors.getOrigins().get(0).toString().contains("example.appspot.com"));
+    assertTrue(cors.getResponseHeaders().contains("Content-Type"));
+    assertEquals(3600, cors.getMaxAgeSeconds().intValue());
+    assertTrue(cors.getMethods().get(0).toString().equalsIgnoreCase("GET"));
     RemoveBucketCors.removeBucketCors(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(RETRY_SETTINGS, () -> assertNull(storage.get(BUCKET).getCors()));
+    assertNull(storage.get(BUCKET).getCors());
   }
 
   @Test
-  public void testRequesterPays() throws Throwable {
+  public void testRequesterPays() throws Exception {
     EnableRequesterPays.enableRequesterPays(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () ->
-            assertTrue(
-                storage
-                    .get(BUCKET, Storage.BucketGetOption.userProject(PROJECT_ID))
-                    .requesterPays()));
+    Bucket bucket = storage.get(BUCKET, Storage.BucketGetOption.userProject(PROJECT_ID));
+    assertTrue(bucket.requesterPays());
     String projectId = ServiceOptions.getDefaultProjectId();
     String blobName = "test-create-empty-blob-requester-pays";
     byte[] content = {0xD, 0xE, 0xA, 0xD};
     Blob remoteBlob =
-        storage.create(
-            BlobInfo.newBuilder(BUCKET, blobName).build(),
-            content,
-            BlobTargetOption.userProject(projectId));
+        bucket.create(blobName, content, Bucket.BlobTargetOption.userProject(projectId));
     assertNotNull(remoteBlob);
     DownloadRequesterPaysObject.downloadRequesterPaysObject(
         projectId, BUCKET, blobName, Paths.get(blobName));
     byte[] readBytes = Files.readAllBytes(Paths.get(blobName));
     assertArrayEquals(content, readBytes);
     DisableRequesterPays.disableRequesterPays(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(RETRY_SETTINGS, () -> assertFalse(storage.get(BUCKET).requesterPays()));
+    assertFalse(storage.get(BUCKET).requesterPays());
   }
 
   @Test
-  public void testRpo() throws Throwable {
+  public void testRpo() throws Exception {
     String rpoBucket = RemoteStorageHelper.generateBucketName();
     try {
       CreateBucketWithTurboReplication.createBucketWithTurboReplication(
           PROJECT_ID, rpoBucket, "NAM4");
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () -> assertEquals("ASYNC_TURBO", storage.get(rpoBucket).getRpo().toString()));
+      Bucket bucket = storage.get(rpoBucket);
+      assertEquals("ASYNC_TURBO", bucket.getRpo().toString());
 
       SetDefaultRpo.setDefaultRpo(PROJECT_ID, rpoBucket);
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () -> assertEquals("DEFAULT", storage.get(rpoBucket).getRpo().toString()));
+      bucket = storage.get(rpoBucket);
+      assertEquals("DEFAULT", bucket.getRpo().toString());
 
       SetAsyncTurboRpo.setAsyncTurboRpo(PROJECT_ID, rpoBucket);
-      TestUtils.retryAssert(
-          RETRY_SETTINGS,
-          () -> assertEquals("ASYNC_TURBO", storage.get(rpoBucket).getRpo().toString()));
+      bucket = storage.get(rpoBucket);
+      assertEquals("ASYNC_TURBO", bucket.getRpo().toString());
 
       GetBucketRpo.getBucketRpo(PROJECT_ID, rpoBucket);
       String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
@@ -603,19 +554,16 @@ public class ITBucketSnippets {
   }
 
   @Test
-  public void testDefaultKMSKey() throws Throwable {
+  public void testDefaultKMSKey() {
     SetBucketDefaultKmsKey.setBucketDefaultKmsKey(PROJECT_ID, BUCKET, KMS_KEY_NAME);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS,
-        () -> assertEquals(KMS_KEY_NAME, storage.get(BUCKET).getDefaultKmsKeyName()));
+    assertEquals(KMS_KEY_NAME, storage.get(BUCKET).getDefaultKmsKeyName());
 
     RemoveBucketDefaultKmsKey.removeBucketDefaultKmsKey(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertNull(storage.get(BUCKET).getDefaultKmsKeyName()));
+    assertNull(storage.get(BUCKET).getDefaultKmsKeyName());
   }
 
   @Test
-  public void testBucketRetention() throws Throwable {
+  public void testBucketRetention() {
     Long retention = 5L;
     SetRetentionPolicy.setRetentionPolicy(PROJECT_ID, BUCKET, retention);
     Bucket bucket = storage.get(BUCKET);
@@ -627,8 +575,7 @@ public class ITBucketSnippets {
     assertTrue(snippetOutput.contains("5"));
 
     EnableDefaultEventBasedHold.enableDefaultEventBasedHold(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertTrue(storage.get(BUCKET).getDefaultEventBasedHold()));
+    assertTrue(storage.get(BUCKET).getDefaultEventBasedHold());
 
     GetDefaultEventBasedHold.getDefaultEventBasedHold(PROJECT_ID, BUCKET);
     snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
@@ -638,23 +585,17 @@ public class ITBucketSnippets {
     String blobName = "test-create-empty-blob-retention-policy";
     bucket.create(blobName, content);
     SetEventBasedHold.setEventBasedHold(PROJECT_ID, BUCKET, blobName);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertTrue(storage.get(BUCKET, blobName).getEventBasedHold()));
+    assertTrue(storage.get(BUCKET, blobName).getEventBasedHold());
     ReleaseEventBasedHold.releaseEventBasedHold(PROJECT_ID, BUCKET, blobName);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertFalse(storage.get(BUCKET, blobName).getEventBasedHold()));
+    assertFalse(storage.get(BUCKET, blobName).getEventBasedHold());
     RemoveRetentionPolicy.removeRetentionPolicy(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertNull(storage.get(BUCKET).getRetentionPeriod()));
+    assertNull(storage.get(BUCKET).getRetentionPeriod());
     DisableDefaultEventBasedHold.disableDefaultEventBasedHold(PROJECT_ID, BUCKET);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertFalse(storage.get(BUCKET).getDefaultEventBasedHold()));
+    assertFalse(storage.get(BUCKET).getDefaultEventBasedHold());
     SetTemporaryHold.setTemporaryHold(PROJECT_ID, BUCKET, blobName);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertTrue(storage.get(BUCKET, blobName).getTemporaryHold()));
+    assertTrue(storage.get(BUCKET, blobName).getTemporaryHold());
     ReleaseTemporaryHold.releaseTemporaryHold(PROJECT_ID, BUCKET, blobName);
-    TestUtils.retryAssert(
-        RETRY_SETTINGS, () -> assertFalse(storage.get(BUCKET, blobName).getTemporaryHold()));
+    assertFalse(storage.get(BUCKET, blobName).getTemporaryHold());
   }
 
   @Test
diff --git a/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java b/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
index 0fb5c058..4c6a1e42 100644
--- a/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
+++ b/samples/snippets/src/test/java/com/example/storage/QuickstartSampleIT.java
@@ -38,14 +38,14 @@ public class QuickstartSampleIT {
 
   private String bucketName;
 
-  private static void deleteBucket(String bucketName) {
+  private static final void deleteBucket(String bucketName) {
     Storage storage = StorageOptions.getDefaultInstance().getService();
     storage.delete(bucketName);
   }
 
   @Before
   public void setUp() {
-    bucketName = "my-new-bucket-" + UUID.randomUUID();
+    bucketName = "my-new-bucket-" + UUID.randomUUID().toString();
   }
 
   @After
@@ -59,18 +59,4 @@ public class QuickstartSampleIT {
     String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
     assertThat(got).contains(String.format("Bucket %s created.", bucketName));
   }
-
-  @Test
-  public void testQuickstartGrpc() throws Exception {
-    QuickstartGrpcSample.main(bucketName);
-    String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
-    assertThat(got).contains(String.format("Bucket %s created.", bucketName));
-  }
-
-  @Test
-  public void testQuickstartGrpcDp() throws Exception {
-    QuickstartGrpcDpSample.main(bucketName);
-    String got = stdOutCaptureRule.getCapturedOutputAsUtf8String();
-    assertThat(got).contains(String.format("Bucket %s created.", bucketName));
-  }
 }
diff --git a/samples/snippets/src/test/java/com/example/storage/TestUtils.java b/samples/snippets/src/test/java/com/example/storage/TestUtils.java
deleted file mode 100644
index 6530c151..00000000
--- a/samples/snippets/src/test/java/com/example/storage/TestUtils.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.example.storage;
-
-import com.google.api.core.CurrentMillisClock;
-import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
-import com.google.api.gax.retrying.RetrySettings;
-import com.google.cloud.RetryHelper;
-import com.google.cloud.RetryHelper.RetryHelperException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.logging.Logger;
-
-public final class TestUtils {
-
-  private static final Logger log = Logger.getLogger(TestUtils.class.getName());
-  private static final Object SENTINEL = new Object();
-
-  private TestUtils() {}
-
-  public static void retryAssert(RetrySettings rs, RetryRunnable f) throws Throwable {
-    AtomicInteger counter = new AtomicInteger(1);
-    try {
-      RetryHelper.runWithRetries(
-          () -> {
-            try {
-              int c = counter.getAndIncrement();
-              if (c > 1) {
-                log.warning(String.format("Retrying assertion for the %d time", c));
-              }
-              f.run();
-              return SENTINEL;
-            } catch (Throwable e) {
-              throw new TunnelThrowable(e);
-            }
-          },
-          rs,
-          new BasicResultRetryAlgorithm<Object>() {
-            @Override
-            public boolean shouldRetry(Throwable previousThrowable, Object previousResponse) {
-              return previousResponse != SENTINEL && previousThrowable != null;
-            }
-          },
-          CurrentMillisClock.getDefaultClock());
-    } catch (RetryHelperException e) {
-      if (e.getCause() instanceof TunnelThrowable) {
-        TunnelThrowable cause = (TunnelThrowable) e.getCause();
-        throw cause.getCause();
-      }
-      throw e.getCause();
-    }
-  }
-
-  @FunctionalInterface
-  public interface RetryRunnable {
-
-    void run() throws Throwable;
-  }
-
-  private static final class TunnelThrowable extends Exception {
-    private TunnelThrowable(Throwable cause) {
-      super(cause);
-    }
-  }
-}
diff --git a/samples/snippets/src/test/java/com/example/storage/TestUtilsTest.java b/samples/snippets/src/test/java/com/example/storage/TestUtilsTest.java
deleted file mode 100644
index d4a76072..00000000
--- a/samples/snippets/src/test/java/com/example/storage/TestUtilsTest.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.example.storage;
-
-import static com.google.common.truth.Truth.assertThat;
-import static org.junit.Assert.assertThrows;
-
-import com.google.api.gax.retrying.RetrySettings;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.junit.Test;
-
-public class TestUtilsTest {
-
-  private static final RetrySettings RETRY_SETTINGS =
-      RetrySettings.newBuilder().setMaxAttempts(3).build();
-
-  @Test
-  public void noException() throws Throwable {
-    TestUtils.retryAssert(RETRY_SETTINGS, () -> assertThat(true).isTrue());
-  }
-
-  @Test
-  public void assertionFailureOnce() throws Throwable {
-    System.out.println("TestUtilsTest.assertionFailureOnce");
-    AtomicInteger c = new AtomicInteger(1);
-    TestUtils.retryAssert(RETRY_SETTINGS, () -> assertThat(c.getAndIncrement()).isGreaterThan(1));
-  }
-
-  @Test
-  public void assertionError_exhausted() throws Throwable {
-    AtomicInteger c = new AtomicInteger(1);
-    try {
-      TestUtils.retryAssert(
-          RETRY_SETTINGS, () -> assertThat(c.getAndIncrement()).isGreaterThan(10));
-      throw new Throwable("expected AssertionError");
-    } catch (AssertionError ignore) {
-      // expected
-    }
-  }
-
-  @Test
-  public void runtimeException_exhausted() {
-    assertThrows(
-        RuntimeException.class,
-        () ->
-            TestUtils.retryAssert(
-                RETRY_SETTINGS,
-                () -> {
-                  throw new RuntimeException("kaboom");
-                }));
-  }
-}