Repository: googleapis/java-storage
Pull request #2059 (2023-06-09 21:12:00)

feat: update Storage.createFrom(BlobInfo, Path) to have 150% higher throughput
When uploading a file where we are able to rewind to an arbitrary offset, we can be more optimistic in the way we send requests to GCS.

Add new code middleware to allow PUTing an entire file to GCS in a single request, and using query resumable session to recover from the specific offset in the case of retryable error.


### Benchmark Results

#### Methodology

Generate a random file on disk of size `128KiB..2GiB` from `/dev/urandom`, then upload the generated file using `Storage.createFrom(BlobInfo, Path)`.

Perform each 4096 times.

Run on a c2-standard-60 instance is us-central1 against a regional bucket located in us-central1.

#### Results

The following summary of throughput in MiB/s as observed between the existing implementation, and the new implementation proposed in this PR.

```
                                count     mean     std    min      50%      75%      90%      99%      max
runId                 ApiName                                                                             
createFrom - existing JSON     4096.0   66.754  10.988  3.249   67.317   73.476   78.961   91.197  107.247
createFrom - new      JSON     4096.0  158.769  67.105  4.600  170.680  218.618  240.992  266.297  305.205
```

#### Comparison

When comparing the new implementation to the existing implementation we get the following improvement to throughput (higher is better):
```
stat      pct
mean  137.841
 50%  153.547
 90%  205.204
 99%  192.003
```



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b4cf5918..c01d9788 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,35 @@
 # Changelog
 
+## [2.24.0](https://github.com/googleapis/java-storage/compare/v2.23.0...v2.24.0) (2023-07-11)
+
+
+### Features
+
+* Add validation around bytes received vs bytes expected ([#2078](https://github.com/googleapis/java-storage/issues/2078)) ([45d142a](https://github.com/googleapis/java-storage/commit/45d142a7ba45a80fca05fc9d8c8d7ce376d3f39c))
+* Initial Preview of Transfer Manager [#2105](https://github.com/googleapis/java-storage/issues/2105) ([8b17574](https://github.com/googleapis/java-storage/commit/8b175745d306e3da58ec3c9307442f14a34fd4ef))
+
+
+### Bug Fixes
+
+* Update grpc upload logic to follow hashing behavior of json ([#2107](https://github.com/googleapis/java-storage/issues/2107)) ([ed05232](https://github.com/googleapis/java-storage/commit/ed05232889c0442f43eacde5dbc80f44f09ebe83))
+* Update UploadCallable to use createFrom to avoid NPE trying to resolve resulting object ([#2086](https://github.com/googleapis/java-storage/issues/2086)) ([6769a2b](https://github.com/googleapis/java-storage/commit/6769a2b3b06fb297ebce819fbdfba5ee377be147))
+
+
+### Documentation
+
+* Javadocs for remainder of Transfer Manager ([#2097](https://github.com/googleapis/java-storage/issues/2097)) ([0362e80](https://github.com/googleapis/java-storage/commit/0362e80386d8055811082204536f596ba948a963))
+* Javadocs for TransferManager interface and ParallelUploadConfig ([#2094](https://github.com/googleapis/java-storage/issues/2094)) ([63d8ed3](https://github.com/googleapis/java-storage/commit/63d8ed37a80304e57b3f48aae614af89a9543ffa))
+* **samples:** Add storage_grpc_quickstart samples ([#2041](https://github.com/googleapis/java-storage/issues/2041)) ([5f916fb](https://github.com/googleapis/java-storage/commit/5f916fbba789edd33c2b4195db885d2b0191f860))
+
+
+### Dependencies
+
+* Update dependency com.google.cloud:google-cloud-shared-dependencies to v3.12.0 ([#2088](https://github.com/googleapis/java-storage/issues/2088)) ([18adb5a](https://github.com/googleapis/java-storage/commit/18adb5a18fc37eb3bb3553fe20648d68a53f987a))
+* Update dependency com.google.cloud:google-cloud-shared-dependencies to v3.13.0 ([#2106](https://github.com/googleapis/java-storage/issues/2106)) ([734dee0](https://github.com/googleapis/java-storage/commit/734dee0402fc16a3cc8d842eb135cc5743586e35))
+* Update dependency io.grpc:grpc-bom to v1.56.1 ([#2101](https://github.com/googleapis/java-storage/issues/2101)) ([3102941](https://github.com/googleapis/java-storage/commit/31029414d8e758c01faaab398a701afebf5e934d))
+* Update dependency net.jqwik:jqwik to v1.7.4 ([#2103](https://github.com/googleapis/java-storage/issues/2103)) ([c681ccf](https://github.com/googleapis/java-storage/commit/c681ccfffe6b861c19b92e496363146adc7d862b))
+* Update gcr.io/cloud-devrel-public-resources/storage-testbench docker tag to v0.36.0 ([#2080](https://github.com/googleapis/java-storage/issues/2080)) ([ba5c11a](https://github.com/googleapis/java-storage/commit/ba5c11a1589d1c7ffefd15e921da914289bc219e))
+
 ## [2.23.0](https://github.com/googleapis/java-storage/compare/v2.22.4...v2.23.0) (2023-06-21)
 
 
diff --git a/gapic-google-cloud-storage-v2/pom.xml b/gapic-google-cloud-storage-v2/pom.xml
index 863c8ff7..cb6b23a3 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.24.1-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.24.1-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..1d0fd5c6 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.24.1-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.24.1-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.24.1-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.24.1-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.24.1-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 9115537c..ccfc4154 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.24.1-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,7 +12,7 @@
   <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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
@@ -248,7 +248,7 @@
     <dependency>
       <groupId>net.jqwik</groupId>
       <artifactId>jqwik</artifactId>
-      <version>1.7.3</version>
+      <version>1.7.4</version>
       <scope>test</scope>
     </dependency>
 
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..524bfe3c 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
@@ -16,6 +16,9 @@
 
 package com.google.cloud.storage;
 
+import static com.google.api.client.util.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkArgument;
+
 import com.google.api.core.InternalApi;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
@@ -124,6 +127,20 @@ abstract class ByteRangeSpec implements Serializable {
     return create(beginOffset, endOffsetInclusive, LeftClosedRightClosedByteRangeSpec::new);
   }
 
+  static ByteRangeSpec parse(String string) {
+    checkNotNull(string, "Range header is null");
+    checkArgument(string.startsWith("bytes="), "malformed Range header value: %s", string);
+
+    int i = string.indexOf('-');
+    String minS = string.substring(6, i);
+    String maxS = string.substring(i + 1);
+
+    long min = Long.parseLong(minS);
+    long max = Long.parseLong(maxS);
+
+    return explicitClosed(min, max);
+  }
+
   private static ByteRangeSpec create(
       @Nullable Long beginOffset,
       @Nullable Long length,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java
index cac56927..98b31a3d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java
@@ -23,10 +23,16 @@ final class ByteSizeConstants {
   static final int _256KiB = 256 * _1KiB;
   static final int _384KiB = 384 * _1KiB;
   static final int _512KiB = 512 * _1KiB;
+  static final int _768KiB = 768 * _1KiB;
   static final int _1MiB = 1024 * _1KiB;
   static final int _2MiB = 2 * _1MiB;
   static final int _16MiB = 16 * _1MiB;
   static final int _32MiB = 32 * _1MiB;
 
+  static final long _128KiBL = 131072L;
+  static final long _256KiBL = 262144L;
+  static final long _512KiBL = 524288L;
+  static final long _768KiBL = 786432L;
+
   private ByteSizeConstants() {}
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index a4164eb2..14b47899 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -103,7 +103,6 @@ import com.google.storage.v2.NotificationConfig;
 import com.google.storage.v2.NotificationConfigName;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ObjectAccessControl;
-import com.google.storage.v2.ObjectChecksums;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.RewriteObjectRequest;
 import com.google.storage.v2.RewriteResponse;
@@ -233,7 +232,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
+    Hasher hasher = Hasher.enabled();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return Retrying.run(
         getOptions(),
@@ -286,8 +285,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
 
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
-
     long size = Files.size(path);
     if (size < bufferSize) {
       // ignore the bufferSize argument if the file is smaller than it
@@ -300,7 +297,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
                 ResumableMedia.gapic()
                     .write()
                     .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge))
-                    .setHasher(hasher)
+                    .setHasher(Hasher.enabled())
                     .setByteStringStrategy(ByteStringStrategy.noCopy())
                     .direct()
                     .buffered(Buffers.allocate(size))
@@ -323,7 +320,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
               .write()
               .byteChannel(
                   storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-              .setHasher(hasher)
+              .setHasher(Hasher.noop())
               .setByteStringStrategy(ByteStringStrategy.noCopy())
               .resumable()
               .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
@@ -359,13 +356,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
     ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
 
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
     BufferedWritableByteChannelSession<WriteObjectResponse> session =
         ResumableMedia.gapic()
             .write()
             .byteChannel(
                 storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(hasher)
+            .setHasher(Hasher.noop())
             .setByteStringStrategy(ByteStringStrategy.noCopy())
             .resumable()
             .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
@@ -739,7 +735,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
-    Hasher hasher = getHasherForRequest(req, Hasher.enabled());
+    Hasher hasher = Hasher.noop();
     return new GrpcBlobWriteChannel(
         storageClient.writeObjectCallable(),
         getOptions(),
@@ -1967,19 +1963,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         Decoder.identity());
   }
 
-  private static Hasher getHasherForRequest(WriteObjectRequest req, Hasher defaultHasher) {
-    if (!req.hasObjectChecksums()) {
-      return defaultHasher;
-    } else {
-      ObjectChecksums checksums = req.getObjectChecksums();
-      if (!checksums.hasCrc32C() && checksums.getMd5Hash().isEmpty()) {
-        return defaultHasher;
-      } else {
-        return Hasher.noop();
-      }
-    }
-  }
-
   @Nullable
   private Blob internalBlobGet(BlobId blob, Opts<ObjectSourceOpt> unwrap) {
     Opts<ObjectSourceOpt> opts = unwrap.resolveFrom(blob).prepend(defaultOpts);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java
new file mode 100644
index 00000000..35c93c02
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java
@@ -0,0 +1,85 @@
+/*
+ * 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;
+
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpRequestFactory;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.util.ObjectParser;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import io.opencensus.trace.Span;
+import io.opencensus.trace.Tracer;
+import io.opencensus.trace.Tracing;
+import java.util.List;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class HttpClientContext {
+
+  private final HttpRequestFactory requestFactory;
+  private final ObjectParser objectParser;
+  private final Tracer tracer;
+
+  private HttpClientContext(
+      HttpRequestFactory requestFactory, ObjectParser objectParser, Tracer tracer) {
+    this.requestFactory = requestFactory;
+    this.objectParser = objectParser;
+    this.tracer = tracer;
+  }
+
+  @SuppressWarnings({"unchecked", "SameParameterValue"})
+  static @Nullable String firstHeaderValue(
+      @NonNull HttpHeaders headers, @NonNull String headerName) {
+    Object v = headers.get(headerName);
+    // HttpHeaders doesn't type its get method, so we have to jump through hoops here
+    if (v instanceof List) {
+      List<String> list = (List<String>) v;
+      return list.get(0);
+    } else {
+      return null;
+    }
+  }
+
+  public HttpRequestFactory getRequestFactory() {
+    return requestFactory;
+  }
+
+  public ObjectParser getObjectParser() {
+    return objectParser;
+  }
+
+  public Tracer getTracer() {
+    return tracer;
+  }
+
+  public Span startSpan(String name) {
+    // record events is hardcoded to true in HttpStorageRpc, preserve it here
+    return tracer.spanBuilder(name).setRecordEvents(true).startSpan();
+  }
+
+  static HttpClientContext from(StorageRpc storageRpc) {
+    return new HttpClientContext(
+        storageRpc.getStorage().getRequestFactory(),
+        storageRpc.getStorage().getObjectParser(),
+        Tracing.getTracer());
+  }
+
+  public static HttpClientContext of(
+      HttpRequestFactory requestFactory, JsonObjectParser jsonObjectParser) {
+    return new HttpClientContext(requestFactory, jsonObjectParser, Tracing.getTracer());
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java
new file mode 100644
index 00000000..8961f523
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+import java.util.function.UnaryOperator;
+
+abstract class HttpContentRange {
+
+  private final boolean finalizing;
+
+  private HttpContentRange(boolean finalizing) {
+    this.finalizing = finalizing;
+  }
+
+  public abstract String getHeaderValue();
+
+  public boolean isFinalizing() {
+    return finalizing;
+  }
+
+  static Total of(ByteRangeSpec spec, long size) {
+    checkArgument(size >= 0, "size must be >= 0");
+    checkArgument(size >= spec.endOffsetInclusive(), "size must be >= end");
+    return new Total(spec, size);
+  }
+
+  static Incomplete of(ByteRangeSpec spec) {
+    return new Incomplete(spec);
+  }
+
+  static Size of(long size) {
+    checkArgument(size >= 0, "size must be >= 0");
+    return new Size(size);
+  }
+
+  static Query query() {
+    return Query.INSTANCE;
+  }
+
+  static HttpContentRange parse(String string) {
+    if ("bytes */*".equals(string)) {
+      return HttpContentRange.query();
+    } else if (string.startsWith("bytes */")) {
+      return HttpContentRange.of(Long.parseLong(string.substring(8)));
+    } else {
+      int idxDash = string.indexOf('-');
+      int idxSlash = string.indexOf('/');
+
+      String beginS = string.substring(6, idxDash);
+      String endS = string.substring(idxDash + 1, idxSlash);
+      long begin = Long.parseLong(beginS);
+      long end = Long.parseLong(endS);
+      if (string.endsWith("/*")) {
+        return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end));
+      } else {
+        String sizeS = string.substring(idxSlash + 1);
+        long size = Long.parseLong(sizeS);
+        return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end), size);
+      }
+    }
+  }
+
+  static final class Incomplete extends HttpContentRange implements HasRange<Incomplete> {
+
+    private final ByteRangeSpec spec;
+
+    private Incomplete(ByteRangeSpec spec) {
+      super(false);
+      this.spec = spec;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes %d-%d/*", spec.beginOffset(), spec.endOffsetInclusive());
+    }
+
+    @Override
+    public ByteRangeSpec range() {
+      return spec;
+    }
+
+    @Override
+    public Incomplete map(UnaryOperator<ByteRangeSpec> f) {
+      return new Incomplete(f.apply(spec));
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Incomplete)) {
+        return false;
+      }
+      Incomplete that = (Incomplete) o;
+      return Objects.equals(spec, that.spec);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(spec);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("spec", spec).toString();
+    }
+  }
+
+  static final class Total extends HttpContentRange implements HasRange<Total>, HasSize {
+
+    private final ByteRangeSpec spec;
+    private final long size;
+
+    private Total(ByteRangeSpec spec, long size) {
+      super(true);
+      this.spec = spec;
+      this.size = size;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes %d-%d/%d", spec.beginOffset(), spec.endOffsetInclusive(), size);
+    }
+
+    @Override
+    public long getSize() {
+      return size;
+    }
+
+    @Override
+    public ByteRangeSpec range() {
+      return spec;
+    }
+
+    @Override
+    public Total map(UnaryOperator<ByteRangeSpec> f) {
+      return new Total(f.apply(spec), size);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Total)) {
+        return false;
+      }
+      Total total = (Total) o;
+      return size == total.size && Objects.equals(spec, total.spec);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(spec, size);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("spec", spec).add("size", size).toString();
+    }
+  }
+
+  static final class Size extends HttpContentRange implements HasSize {
+
+    private final long size;
+
+    private Size(long size) {
+      super(true);
+      this.size = size;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes */%d", size);
+    }
+
+    @Override
+    public long getSize() {
+      return size;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Size)) {
+        return false;
+      }
+      Size size1 = (Size) o;
+      return size == size1.size;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(size);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("size", size).toString();
+    }
+  }
+
+  static final class Query extends HttpContentRange {
+
+    private static final Query INSTANCE = new Query();
+
+    private Query() {
+      super(false);
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return "bytes */*";
+    }
+  }
+
+  interface HasRange<T extends HttpContentRange> {
+
+    ByteRangeSpec range();
+
+    T map(UnaryOperator<ByteRangeSpec> f);
+  }
+
+  interface HasSize {
+
+    long getSize();
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
index 3483dd90..b525cf75 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
@@ -29,6 +29,7 @@ import com.google.cloud.ServiceRpc;
 import com.google.cloud.TransportOptions;
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.spi.ServiceRpcFactory;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.cloud.storage.spi.v1.HttpStorageRpc;
@@ -51,6 +52,7 @@ public class HttpStorageOptions extends StorageOptions {
   private static final String DEFAULT_HOST = "https://storage.googleapis.com";
 
   private final HttpRetryAlgorithmManager retryAlgorithmManager;
+  private final transient RetryDependenciesAdapter retryDepsAdapter;
 
   private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
@@ -58,6 +60,7 @@ public class HttpStorageOptions extends StorageOptions {
         new HttpRetryAlgorithmManager(
             MoreObjects.firstNonNull(
                 builder.storageRetryStrategy, defaults().getStorageRetryStrategy()));
+    retryDepsAdapter = new RetryDependenciesAdapter();
   }
 
   @Override
@@ -102,6 +105,11 @@ public class HttpStorageOptions extends StorageOptions {
     return HttpStorageDefaults.INSTANCE;
   }
 
+  @InternalApi
+  RetryingDependencies asRetryDependencies() {
+    return retryDepsAdapter;
+  }
+
   public static class Builder extends StorageOptions.Builder {
 
     private StorageRetryStrategy storageRetryStrategy;
@@ -321,4 +329,23 @@ public class HttpStorageOptions extends StorageOptions {
       }
     }
   }
+
+  /**
+   * We don't yet want to make HttpStorageOptions itself implement {@link RetryingDependencies} but
+   * we do need use it in a couple places, for those we create this adapter.
+   */
+  private final class RetryDependenciesAdapter implements RetryingDependencies {
+
+    private RetryDependenciesAdapter() {}
+
+    @Override
+    public RetrySettings getRetrySettings() {
+      return HttpStorageOptions.this.getRetrySettings();
+    }
+
+    @Override
+    public ApiClock getClock() {
+      return HttpStorageOptions.this.getClock();
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java
new file mode 100644
index 00000000..f5935550
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java
@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.spi.v1.HttpStorageRpc;
+import io.opencensus.trace.EndSpanOptions;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class JsonResumableSession extends ResumableSession<StorageObject> {
+
+  static final String SPAN_NAME_WRITE =
+      String.format("Sent.%s.write", HttpStorageRpc.class.getName());
+  static final EndSpanOptions END_SPAN_OPTIONS =
+      EndSpanOptions.builder().setSampleToLocalSpanStore(true).build();
+
+  private final HttpClientContext context;
+  private final RetryingDependencies deps;
+  private final ResultRetryAlgorithm<?> alg;
+  private final JsonResumableWrite resumableWrite;
+
+  JsonResumableSession(
+      HttpClientContext context,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg,
+      JsonResumableWrite resumableWrite) {
+    this.context = context;
+    this.deps = deps;
+    this.alg = alg;
+    this.resumableWrite = resumableWrite;
+  }
+
+  /**
+   * Not automatically retried. Usually called from within another retrying context. We don't yet
+   * have the concept of nested retry handling.
+   */
+  @Override
+  ResumableOperationResult<@Nullable StorageObject> query() {
+    return new JsonResumableSessionQueryTask(context, resumableWrite.getUploadId()).call();
+  }
+
+  @Override
+  ResumableOperationResult<@Nullable StorageObject> put(
+      RewindableHttpContent content, HttpContentRange contentRange) {
+    JsonResumableSessionPutTask task =
+        new JsonResumableSessionPutTask(
+            context, resumableWrite.getUploadId(), content, contentRange);
+    AtomicBoolean dirty = new AtomicBoolean(false);
+    return Retrying.run(
+        deps,
+        alg,
+        () -> {
+          if (dirty.getAndSet(true)) {
+            ResumableOperationResult<@Nullable StorageObject> query = query();
+            if (query.getObject() != null) {
+              return query;
+            } else {
+              task.rewindTo(query.getPersistedSize());
+            }
+          }
+          return task.call();
+        },
+        Decoder.identity());
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java
new file mode 100644
index 00000000..2b6e8d56
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java
@@ -0,0 +1,233 @@
+/*
+ * 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;
+
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpResponse;
+import com.google.api.client.http.HttpResponseException;
+import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.StorageException.IOExceptionCallable;
+import com.google.common.io.CharStreams;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.function.Predicate;
+import javax.annotation.ParametersAreNonnullByDefault;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@ParametersAreNonnullByDefault
+enum JsonResumableSessionFailureScenario {
+  // TODO: send more bytes than are in the Content-Range header
+  SCENARIO_0(BaseServiceException.UNKNOWN_CODE, null, "Unknown Error"),
+  SCENARIO_0_1(BaseServiceException.UNKNOWN_CODE, null, "Response not application/json."),
+  SCENARIO_1(
+      BaseServiceException.UNKNOWN_CODE,
+      "invalid",
+      "Attempt to append to already finalized resumable session."),
+  SCENARIO_2(
+      BaseServiceException.UNKNOWN_CODE,
+      "invalid",
+      "Attempt to finalize resumable session with fewer bytes than the backend has received."),
+  SCENARIO_3(
+      BaseServiceException.UNKNOWN_CODE,
+      "dataLoss",
+      "Attempt to finalize resumable session with more bytes than the backend has received."),
+  SCENARIO_4(200, "ok", "Attempt to finalize an already finalized session with same object size"),
+  SCENARIO_4_1(
+      BaseServiceException.UNKNOWN_CODE,
+      "dataLoss",
+      "Finalized resumable session, but object size less than expected."),
+  SCENARIO_4_2(
+      BaseServiceException.UNKNOWN_CODE,
+      "dataLoss",
+      "Finalized resumable session, but object size greater than expected."),
+  SCENARIO_5(
+      BaseServiceException.UNKNOWN_CODE,
+      "dataLoss",
+      "Client side data loss detected. Attempt to append to a resumable session with an offset higher than the backend has"),
+  SCENARIO_7(
+      BaseServiceException.UNKNOWN_CODE,
+      "dataLoss",
+      "Client side data loss detected. Bytes acked is more than client sent."),
+  SCENARIO_9(503, "backendNotConnected", "Ack less than bytes sent"),
+  QUERY_SCENARIO_1(503, "", "Missing Range header in response");
+
+  private static final String PREFIX_I = "\t|< ";
+  private static final String PREFIX_O = "\t|> ";
+  private static final String PREFIX_X = "\t|  ";
+
+  private static final Predicate<String> includedHeaders =
+      matches("Content-Length")
+          .or(matches("Content-Encoding"))
+          .or(matches("Content-Range"))
+          .or(matches("Content-Type"))
+          .or(matches("Range"))
+          .or(startsWith("X-Goog-Stored-"))
+          .or(matches("X-GUploader-UploadID"));
+
+  private static final Predicate<Map.Entry<String, ?>> includeHeader =
+      e -> includedHeaders.test(e.getKey());
+
+  private final int code;
+  @Nullable private final String reason;
+  private final String message;
+
+  JsonResumableSessionFailureScenario(int code, @Nullable String reason, String message) {
+    this.code = code;
+    this.reason = reason;
+    this.message = message;
+  }
+
+  StorageException toStorageException(String uploadId, HttpResponse resp) {
+    return toStorageException(
+        uploadId, resp, null, () -> CharStreams.toString(new InputStreamReader(resp.getContent())));
+  }
+
+  StorageException toStorageException(
+      String uploadId, @Nullable HttpResponse resp, @Nullable Throwable cause) {
+    if (resp != null) {
+      // an exception caused this, do not try to read the content from the response.
+      return toStorageException(uploadId, resp, cause, () -> null);
+    } else {
+      return new StorageException(code, message, reason, cause);
+    }
+  }
+
+  StorageException toStorageException(
+      String uploadId,
+      HttpResponse resp,
+      @Nullable Throwable cause,
+      IOExceptionCallable<@Nullable String> contentCallable) {
+    return toStorageException(code, message, reason, uploadId, resp, cause, contentCallable);
+  }
+
+  static StorageException toStorageException(
+      HttpResponse response, HttpResponseException cause, String uploadId) {
+    String statusMessage = cause.getStatusMessage();
+    StorageException se =
+        JsonResumableSessionFailureScenario.toStorageException(
+            cause.getStatusCode(),
+            String.format(
+                "%d %s", cause.getStatusCode(), statusMessage == null ? "" : statusMessage),
+            "",
+            uploadId,
+            response,
+            cause,
+            () -> null);
+    return se;
+  }
+
+  static StorageException toStorageException(
+      int overrideCode,
+      String message,
+      @Nullable String reason,
+      String uploadId,
+      HttpResponse resp,
+      @Nullable Throwable cause,
+      IOExceptionCallable<@Nullable String> contentCallable) {
+    Throwable suppress = null;
+    StringBuilder sb = new StringBuilder();
+    sb.append(message);
+    // add request context
+    sb.append("\n").append(PREFIX_O).append("PUT ").append(uploadId);
+    recordHeaderTo(resp.getRequest().getHeaders(), PREFIX_O, sb);
+
+    sb.append("\n").append(PREFIX_X);
+    // add response context
+    {
+      int code = resp.getStatusCode();
+      sb.append("\n").append(PREFIX_I).append("HTTP/1.1 ").append(code);
+      if (resp.getStatusMessage() != null) {
+        sb.append(" ").append(resp.getStatusMessage());
+      }
+
+      recordHeaderTo(resp.getHeaders(), PREFIX_I, sb);
+      // try to include any body that we can handle
+      if (isOk(code) || code == 503 || code == 400) {
+        try {
+          String content = contentCallable.call();
+          if (content != null) {
+            sb.append("\n").append(PREFIX_I);
+            if (content.contains("\n") || content.contains("\r\n")) {
+              sb.append("\n").append(PREFIX_I).append(content.replaceAll("\r?\n", "\n" + PREFIX_I));
+            } else {
+              sb.append("\n").append(PREFIX_I).append(content);
+            }
+          }
+        } catch (IOException e) {
+          // com.google.api.client.http.HttpResponseException.Builder.Builder
+          // prints an exception which might occur while attempting to resolve the content
+          // this can lose the context about the request it was for, instead we register it
+          // as a suppressed exception
+          suppress = new StorageException(0, "Error reading response content for diagnostics.", e);
+        }
+      }
+
+      sb.append("\n").append(PREFIX_X);
+    }
+    StorageException storageException =
+        new StorageException(overrideCode, sb.toString(), reason, cause);
+    if (suppress != null) {
+      storageException.addSuppressed(suppress);
+    }
+    return storageException;
+  }
+
+  static boolean isOk(int code) {
+    return code == 200 || code == 201;
+  }
+
+  static boolean isContinue(int code) {
+    return code == 308;
+  }
+
+  // The header names from HttpHeaders are lower cased, define some utility methods to create
+  // predicates where we can specify values ignoring case
+  private static Predicate<String> matches(String expected) {
+    String lower = expected.toLowerCase(Locale.US);
+    return lower::equals;
+  }
+
+  private static Predicate<String> startsWith(String prefix) {
+    String lower = prefix.toLowerCase(Locale.US);
+    return s -> s.startsWith(lower);
+  }
+
+  private static void recordHeaderTo(HttpHeaders h, String prefix, StringBuilder sb) {
+    h.entrySet().stream()
+        .filter(includeHeader)
+        .forEach(
+            e -> {
+              String key = e.getKey();
+              String value = headerValueToString(e.getValue());
+              sb.append("\n").append(prefix).append(key).append(": ").append(value);
+            });
+  }
+
+  private static String headerValueToString(Object o) {
+    if (o instanceof List) {
+      List<?> l = (List<?>) o;
+      if (l.size() == 1) {
+        return l.get(0).toString();
+      }
+    }
+
+    return o.toString();
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
new file mode 100644
index 00000000..73b7d14a
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
@@ -0,0 +1,218 @@
+/*
+ * 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;
+
+import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpRequest;
+import com.google.api.client.http.HttpResponse;
+import com.google.api.client.http.HttpResponseException;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.StorageException.IOExceptionCallable;
+import com.google.common.annotations.VisibleForTesting;
+import io.opencensus.common.Scope;
+import io.opencensus.trace.Span;
+import io.opencensus.trace.Status;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.Locale;
+import java.util.concurrent.Callable;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class JsonResumableSessionPutTask
+    implements Callable<ResumableOperationResult<@Nullable StorageObject>> {
+
+  private final HttpClientContext context;
+  private final String uploadId;
+  private final RewindableHttpContent content;
+  private final HttpContentRange originalContentRange;
+
+  private HttpContentRange contentRange;
+
+  @VisibleForTesting
+  JsonResumableSessionPutTask(
+      HttpClientContext httpClientContext,
+      String uploadId,
+      RewindableHttpContent content,
+      HttpContentRange originalContentRange) {
+    this.context = httpClientContext;
+    this.uploadId = uploadId;
+    this.content = content;
+    this.originalContentRange = originalContentRange;
+    this.contentRange = originalContentRange;
+  }
+
+  public void rewindTo(long offset) {
+    content.rewindTo(offset);
+    if (contentRange instanceof HttpContentRange.HasRange) {
+      HttpContentRange.HasRange<?> range = (HttpContentRange.HasRange<?>) contentRange;
+      contentRange = range.map(s -> s.withNewBeginOffset(offset));
+    }
+  }
+
+  public ResumableOperationResult<@Nullable StorageObject> call() throws IOException {
+    Span span = context.startSpan(JsonResumableSession.SPAN_NAME_WRITE);
+    Scope scope = context.getTracer().withSpan(span);
+
+    boolean success = false;
+    boolean finalizing = originalContentRange.isFinalizing();
+
+    HttpRequest req =
+        context
+            .getRequestFactory()
+            .buildPutRequest(new GenericUrl(uploadId), content)
+            .setParser(context.getObjectParser());
+    req.setThrowExceptionOnExecuteError(false);
+    req.getHeaders().setContentRange(contentRange.getHeaderValue());
+
+    HttpResponse response = null;
+    try {
+      response = req.execute();
+
+      int code = response.getStatusCode();
+
+      if (!finalizing && JsonResumableSessionFailureScenario.isContinue(code)) {
+        long effectiveEnd = ((HttpContentRange.HasRange<?>) contentRange).range().endOffset();
+        @Nullable String range = response.getHeaders().getRange();
+        ByteRangeSpec ackRange = ByteRangeSpec.parse(range);
+        if (ackRange.endOffset() == effectiveEnd) {
+          success = true;
+          return ResumableOperationResult.incremental(ackRange.endOffset());
+        } else if (ackRange.endOffset() < effectiveEnd) {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_9.toStorageException(uploadId, response);
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        } else {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        }
+      } else if (finalizing && JsonResumableSessionFailureScenario.isOk(code)) {
+        @Nullable StorageObject storageObject;
+        @Nullable BigInteger actualSize;
+
+        Long contentLength = response.getHeaders().getContentLength();
+        String contentType = response.getHeaders().getContentType();
+        String storedContentLength =
+            HttpClientContext.firstHeaderValue(
+                response.getHeaders(), "x-goog-stored-content-length");
+        boolean isJson = contentType != null && contentType.startsWith("application/json");
+        if (isJson) {
+          storageObject = response.parseAs(StorageObject.class);
+          actualSize = storageObject != null ? storageObject.getSize() : null;
+        } else if ((contentLength == null || contentLength == 0) && storedContentLength != null) {
+          // when a signed url is used, the finalize response is empty
+          response.ignore();
+          actualSize = new BigInteger(storedContentLength, 10);
+          success = true;
+          storageObject = null;
+        } else {
+          response.ignore();
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
+                  uploadId, response, null, () -> null);
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        }
+        BigInteger expectedSize =
+            BigInteger.valueOf(((HttpContentRange.HasSize) contentRange).getSize());
+        int compare = expectedSize.compareTo(actualSize);
+        if (compare == 0) {
+          success = true;
+          //noinspection DataFlowIssue  compareTo result will filter out actualSize == null
+          return ResumableOperationResult.complete(storageObject, actualSize.longValue());
+        } else if (compare < 0) {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                  uploadId, response, null, toString(storageObject));
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        } else {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                  uploadId, response, null, toString(storageObject));
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        }
+      } else if (!finalizing && JsonResumableSessionFailureScenario.isOk(code)) {
+        StorageException se =
+            JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException(uploadId, response);
+        span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+        throw se;
+      } else if (finalizing && JsonResumableSessionFailureScenario.isContinue(code)) {
+        // in order to finalize the content range must have a size, cast down to read it
+        HttpContentRange.HasSize size = (HttpContentRange.HasSize) contentRange;
+
+        ByteRangeSpec range = ByteRangeSpec.parse(response.getHeaders().getRange());
+        if (range.endOffsetInclusive() < size.getSize()) {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_3.toStorageException(uploadId, response);
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        } else {
+          StorageException se =
+              JsonResumableSessionFailureScenario.SCENARIO_2.toStorageException(uploadId, response);
+          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+          throw se;
+        }
+      } else {
+        HttpResponseException cause = new HttpResponseException(response);
+        String contentType = response.getHeaders().getContentType();
+        // If the content-range header value has run ahead of the backend, it will respond with
+        // a 503 with plain text content
+        // Attempt to detect this very loosely as to minimize impact of modified error message
+        // This is accurate circa 2023-06
+        if ((!JsonResumableSessionFailureScenario.isOk(code)
+                && !JsonResumableSessionFailureScenario.isContinue(code))
+            && contentType != null
+            && contentType.startsWith("text/plain")) {
+          String errorMessage = cause.getContent().toLowerCase(Locale.US);
+          if (errorMessage.contains("content-range")) {
+            StorageException se =
+                JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                    uploadId, response, cause, cause::getContent);
+            span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+            throw se;
+          }
+        }
+        StorageException se =
+            JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
+        span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+        throw se;
+      }
+    } catch (StorageException e) {
+      span.setStatus(Status.UNKNOWN.withDescription(e.getMessage()));
+      throw e;
+    } catch (Exception e) {
+      StorageException se =
+          JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(uploadId, response, e);
+      span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
+      throw se;
+    } finally {
+      if (success && !finalizing && response != null) {
+        response.ignore();
+      }
+      scope.close();
+      span.end(JsonResumableSession.END_SPAN_OPTIONS);
+    }
+  }
+
+  static IOExceptionCallable<@Nullable String> toString(@Nullable Object o) {
+    return () -> o != null ? o.toString() : null;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
new file mode 100644
index 00000000..b37d2396
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
@@ -0,0 +1,136 @@
+/*
+ * 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;
+
+import static com.google.cloud.storage.HttpClientContext.firstHeaderValue;
+
+import com.google.api.client.http.EmptyContent;
+import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpRequest;
+import com.google.api.client.http.HttpResponse;
+import com.google.api.client.http.HttpResponseException;
+import com.google.api.services.storage.model.StorageObject;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.Locale;
+import java.util.concurrent.Callable;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class JsonResumableSessionQueryTask
+    implements Callable<ResumableOperationResult<@Nullable StorageObject>> {
+
+  private final HttpClientContext context;
+  private final String uploadId;
+
+  JsonResumableSessionQueryTask(HttpClientContext context, String uploadId) {
+    this.context = context;
+    this.uploadId = uploadId;
+  }
+
+  public ResumableOperationResult<@Nullable StorageObject> call() {
+    HttpResponse response = null;
+    try {
+      HttpRequest req =
+          context
+              .getRequestFactory()
+              .buildPutRequest(new GenericUrl(uploadId), new EmptyContent())
+              .setParser(context.getObjectParser());
+      req.setThrowExceptionOnExecuteError(false);
+      req.getHeaders().setContentRange(HttpContentRange.query().getHeaderValue());
+
+      response = req.execute();
+
+      int code = response.getStatusCode();
+      if (JsonResumableSessionFailureScenario.isOk(code)) {
+        @Nullable StorageObject storageObject;
+        @Nullable BigInteger actualSize;
+
+        Long contentLength = response.getHeaders().getContentLength();
+        String contentType = response.getHeaders().getContentType();
+        String storedContentLength =
+            firstHeaderValue(response.getHeaders(), "x-goog-stored-content-length");
+        boolean isJson = contentType != null && contentType.startsWith("application/json");
+        if (isJson) {
+          storageObject = response.parseAs(StorageObject.class);
+          actualSize = storageObject != null ? storageObject.getSize() : null;
+        } else if ((contentLength == null || contentLength == 0) && storedContentLength != null) {
+          // when a signed url is used, the finalize response is empty
+          response.ignore();
+          actualSize = new BigInteger(storedContentLength, 10);
+          storageObject = null;
+        } else {
+          response.ignore();
+          throw JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
+              uploadId, response, null, () -> null);
+        }
+        if (actualSize != null) {
+          if (storageObject != null) {
+            return ResumableOperationResult.complete(storageObject, actualSize.longValue());
+          } else {
+            return ResumableOperationResult.incremental(actualSize.longValue());
+          }
+        } else {
+          throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+              uploadId,
+              response,
+              null,
+              () -> storageObject != null ? storageObject.toString() : null);
+        }
+      } else if (JsonResumableSessionFailureScenario.isContinue(code)) {
+        String range1 = response.getHeaders().getRange();
+        if (range1 != null) {
+          ByteRangeSpec range = ByteRangeSpec.parse(range1);
+          long endOffset = range.endOffset();
+          return ResumableOperationResult.incremental(endOffset);
+        } else {
+          throw JsonResumableSessionFailureScenario.QUERY_SCENARIO_1.toStorageException(
+              uploadId, response);
+        }
+      } else {
+        HttpResponseException cause = new HttpResponseException(response);
+        String contentType = response.getHeaders().getContentType();
+        // If the content-range header value has run ahead of the backend, it will respond with
+        // a 503 with plain text content
+        // Attempt to detect this very loosely as to minimize impact of modified error message
+        // This is accurate circa 2023-06
+        if ((!JsonResumableSessionFailureScenario.isOk(code)
+                && !JsonResumableSessionFailureScenario.isContinue(code))
+            && contentType != null
+            && contentType.startsWith("text/plain")) {
+          String errorMessage = cause.getContent().toLowerCase(Locale.US);
+          if (errorMessage.contains("content-range")) {
+            throw JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                uploadId, response, cause, cause::getContent);
+          }
+        }
+        throw JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
+      }
+    } catch (StorageException se) {
+      throw se;
+    } catch (Exception e) {
+      throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+          uploadId, response, e);
+    } finally {
+      if (response != null) {
+        try {
+          response.ignore();
+        } catch (IOException ignore) {
+        }
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java
new file mode 100644
index 00000000..c5e17aad
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java
@@ -0,0 +1,88 @@
+/*
+ * 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;
+
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.base.MoreObjects;
+import java.util.Map;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class JsonResumableWrite {
+  @MonotonicNonNull private final StorageObject object;
+  @MonotonicNonNull private final Map<StorageRpc.Option, ?> options;
+
+  @MonotonicNonNull private final String signedUrl;
+
+  @NonNull private final String uploadId;
+
+  private JsonResumableWrite(
+      StorageObject object,
+      Map<StorageRpc.Option, ?> options,
+      String signedUrl,
+      @NonNull String uploadId) {
+    this.object = object;
+    this.options = options;
+    this.signedUrl = signedUrl;
+    this.uploadId = uploadId;
+  }
+
+  public @NonNull String getUploadId() {
+    return uploadId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof JsonResumableWrite)) {
+      return false;
+    }
+    JsonResumableWrite that = (JsonResumableWrite) o;
+    return Objects.equals(object, that.object)
+        && Objects.equals(options, that.options)
+        && Objects.equals(signedUrl, that.signedUrl)
+        && uploadId.equals(that.uploadId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(object, options, signedUrl, uploadId);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("object", object)
+        .add("options", options)
+        .add("signedUrl", signedUrl)
+        .add("uploadId", uploadId)
+        .toString();
+  }
+
+  static JsonResumableWrite of(
+      StorageObject req, Map<StorageRpc.Option, ?> options, String uploadId) {
+    return new JsonResumableWrite(req, options, null, uploadId);
+  }
+
+  static JsonResumableWrite of(String signedUrl, String uploadId) {
+    return new JsonResumableWrite(null, null, signedUrl, uploadId);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java
new file mode 100644
index 00000000..88b5b756
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java
@@ -0,0 +1,90 @@
+/*
+ * 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;
+
+import com.google.common.base.MoreObjects;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+abstract class ResumableOperationResult<@Nullable T> {
+
+  private ResumableOperationResult() {}
+
+  abstract @Nullable T getObject();
+
+  abstract long getPersistedSize();
+
+  static <T> ResumableOperationResult<T> complete(T t, long persistedSize) {
+    return new CompletedResult<>(t, persistedSize);
+  }
+
+  static <@Nullable T> ResumableOperationResult<T> incremental(long persistedSize) {
+    return new IncrementalResult<>(persistedSize);
+  }
+
+  private static final class CompletedResult<T> extends ResumableOperationResult<T> {
+
+    private final long persistedSize;
+    private final T entity;
+
+    private CompletedResult(T entity, long persistedSize) {
+      this.entity = entity;
+      this.persistedSize = persistedSize;
+    }
+
+    @Override
+    public @Nullable T getObject() {
+      return entity;
+    }
+
+    @Override
+    public long getPersistedSize() {
+      return persistedSize;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("persistedSize", persistedSize)
+          .add("entity", entity)
+          .toString();
+    }
+  }
+
+  private static final class IncrementalResult<@Nullable T> extends ResumableOperationResult<T> {
+
+    private final long persistedSize;
+
+    private IncrementalResult(long persistedSize) {
+      this.persistedSize = persistedSize;
+    }
+
+    @Override
+    public @Nullable T getObject() {
+      return null;
+    }
+
+    @Override
+    public long getPersistedSize() {
+      return persistedSize;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("persistedSize", persistedSize).toString();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java
new file mode 100644
index 00000000..7b608c68
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.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;
+
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+abstract class ResumableSession<T> {
+
+  ResumableSession() {}
+
+  abstract ResumableOperationResult<@Nullable T> put(
+      RewindableHttpContent content, HttpContentRange contentRange);
+
+  abstract ResumableOperationResult<@Nullable T> query();
+
+  static JsonResumableSession json(
+      HttpClientContext context,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg,
+      JsonResumableWrite resumableWrite) {
+    return new JsonResumableSession(context, deps, alg, resumableWrite);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java
new file mode 100644
index 00000000..fd171e05
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java
@@ -0,0 +1,106 @@
+/*
+ * 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;
+
+import com.google.api.client.http.AbstractHttpContent;
+import com.google.api.client.http.HttpMediaType;
+import com.google.common.base.Preconditions;
+import com.google.common.io.ByteStreams;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+abstract class RewindableHttpContent extends AbstractHttpContent {
+
+  private RewindableHttpContent() {
+    super((HttpMediaType) null);
+  }
+
+  @Override
+  public abstract long getLength();
+
+  abstract void rewindTo(long offset);
+
+  @Override
+  public final boolean retrySupported() {
+    return false;
+  }
+
+  static RewindableHttpContent empty() {
+    return EmptyRewindableContent.INSTANCE;
+  }
+
+  static RewindableHttpContent of(Path path) throws IOException {
+    return new PathRewindableHttpContent(path);
+  }
+
+  private static final class EmptyRewindableContent extends RewindableHttpContent {
+    private static final EmptyRewindableContent INSTANCE = new EmptyRewindableContent();
+
+    @Override
+    public long getLength() {
+      return 0L;
+    }
+
+    @Override
+    public void writeTo(OutputStream out) throws IOException {
+      out.flush();
+    }
+
+    @Override
+    protected void rewindTo(long offset) {}
+  }
+
+  private static final class PathRewindableHttpContent extends RewindableHttpContent {
+
+    private final Path path;
+    private final long size;
+
+    private long readOffset;
+
+    private PathRewindableHttpContent(Path path) throws IOException {
+      this.path = path;
+      this.size = Files.size(path);
+      this.readOffset = 0;
+    }
+
+    @Override
+    public long getLength() {
+      return size - readOffset;
+    }
+
+    @Override
+    protected void rewindTo(long offset) {
+      Preconditions.checkArgument(
+          offset < size, "provided offset must be less than size (%d < %d)", offset, size);
+      this.readOffset = offset;
+    }
+
+    @Override
+    public void writeTo(OutputStream out) throws IOException {
+      try (SeekableByteChannel in = Files.newByteChannel(path, StandardOpenOption.READ)) {
+        in.position(readOffset);
+        ByteStreams.copy(in, Channels.newChannel(out));
+        out.flush();
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index 8c0768b9..a97a325c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -86,6 +86,8 @@ import java.util.TimeZone;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.Supplier;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
@@ -225,9 +227,40 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     if (Files.isDirectory(path)) {
       throw new StorageException(0, path + " is a directory");
     }
-    try (InputStream input = Files.newInputStream(path)) {
-      return createFrom(blobInfo, input, bufferSize, options);
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
+    final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+    BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
+    BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+    StorageObject encode = codecs.blobInfo().encode(updated);
+
+    Supplier<String> uploadIdSupplier =
+        ResumableMedia.startUploadForBlobInfo(
+            getOptions(),
+            updated,
+            optionsMap,
+            retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap));
+    JsonResumableWrite jsonResumableWrite =
+        JsonResumableWrite.of(encode, optionsMap, uploadIdSupplier.get());
+
+    JsonResumableSession session =
+        ResumableSession.json(
+            HttpClientContext.from(storageRpc),
+            getOptions().asRetryDependencies(),
+            retryAlgorithmManager.idempotent(),
+            jsonResumableWrite);
+    long size = Files.size(path);
+    HttpContentRange contentRange =
+        HttpContentRange.of(ByteRangeSpec.relativeLength(0L, size), size);
+    ResumableOperationResult<StorageObject> put =
+        session.put(RewindableHttpContent.of(path), contentRange);
+    // all exception translation is taken care of down in the JsonResumableSession
+    StorageObject object = put.getObject();
+    if (object == null) {
+      // if by some odd chance the put didn't get the StorageObject, query for it
+      ResumableOperationResult<@Nullable StorageObject> query = session.query();
+      object = query.getObject();
     }
+    return codecs.blobInfo().decode(object).asBlob(this);
   }
 
   @Override
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..e51dfb90 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.24.1-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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/pom.xml b/pom.xml
index cebc4695..af3f0ffd 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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -54,7 +54,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <github.global.server>github</github.global.server>
     <site.installationModule>google-cloud-storage-parent</site.installationModule>
-    <google.cloud.shared-dependencies.version>3.12.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>3.13.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <dependencyManagement>
@@ -62,7 +62,7 @@
       <dependency>
         <groupId>io.grpc</groupId>
         <artifactId>grpc-bom</artifactId>
-        <version>1.56.0</version>
+        <version>1.56.1</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -83,7 +83,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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.apis</groupId>
@@ -124,17 +124,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.24.1-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.24.1-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.24.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
diff --git a/proto-google-cloud-storage-v2/pom.xml b/proto-google-cloud-storage-v2/pom.xml
index 1c6a5e2d..9068b9ed 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.24.1-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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 0127f645..3723a7f2 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.24.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
 
     <dependency>
diff --git a/versions.txt b/versions.txt
index 819d5aaf..1cd08c23 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.24.0:2.24.1-SNAPSHOT
+gapic-google-cloud-storage-v2:2.24.0-alpha:2.24.1-alpha-SNAPSHOT
+grpc-google-cloud-storage-v2:2.24.0-alpha:2.24.1-alpha-SNAPSHOT
+proto-google-cloud-storage-v2:2.24.0-alpha:2.24.1-alpha-SNAPSHOT


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java
new file mode 100644
index 00000000..be739ab1
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java
@@ -0,0 +1,144 @@
+/*
+ * 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;
+
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONNECTION;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderValues.CLOSE;
+
+import io.grpc.netty.shaded.io.netty.bootstrap.ServerBootstrap;
+import io.grpc.netty.shaded.io.netty.buffer.ByteBuf;
+import io.grpc.netty.shaded.io.netty.channel.Channel;
+import io.grpc.netty.shaded.io.netty.channel.ChannelFuture;
+import io.grpc.netty.shaded.io.netty.channel.ChannelFutureListener;
+import io.grpc.netty.shaded.io.netty.channel.ChannelHandlerContext;
+import io.grpc.netty.shaded.io.netty.channel.ChannelInitializer;
+import io.grpc.netty.shaded.io.netty.channel.ChannelOption;
+import io.grpc.netty.shaded.io.netty.channel.ChannelPipeline;
+import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.SimpleChannelInboundHandler;
+import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.socket.SocketChannel;
+import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaders;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpRequest;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpServerCodec;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpServerExpectContinueHandler;
+import io.grpc.netty.shaded.io.netty.handler.logging.LogLevel;
+import io.grpc.netty.shaded.io.netty.handler.logging.LoggingHandler;
+import java.net.InetSocketAddress;
+import java.net.URI;
+
+final class FakeHttpServer implements AutoCloseable {
+
+  private final URI endpoint;
+  private final Channel channel;
+  private final Runnable shutdown;
+
+  private FakeHttpServer(URI endpoint, Channel channel, Runnable shutdown) {
+    this.endpoint = endpoint;
+    this.channel = channel;
+    this.shutdown = shutdown;
+  }
+
+  public URI getEndpoint() {
+    return endpoint;
+  }
+
+  @Override
+  public void close() throws Exception {
+    shutdown.run();
+    channel.closeFuture().syncUninterruptibly();
+  }
+
+  static FakeHttpServer of(HttpRequestHandler server) {
+    // based on
+    // https://github.com/netty/netty/blob/59aa6e635b9996cf21cd946e64353270679adc73/example/src/main/java/io/netty/example/http/helloworld/HttpHelloWorldServer.java
+    InetSocketAddress address = new InetSocketAddress("localhost", 0);
+    // Configure the server.
+    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
+    EventLoopGroup workerGroup = new NioEventLoopGroup();
+    ServerBootstrap b = new ServerBootstrap();
+    b.option(ChannelOption.SO_BACKLOG, 1024);
+    b.group(bossGroup, workerGroup)
+        .channel(NioServerSocketChannel.class)
+        .handler(new LoggingHandler(LogLevel.DEBUG))
+        .childHandler(
+            new ChannelInitializer<SocketChannel>() {
+              @Override
+              protected void initChannel(SocketChannel ch) {
+                ChannelPipeline p = ch.pipeline();
+                p.addLast(new HttpServerCodec());
+                p.addLast(new HttpServerExpectContinueHandler());
+                p.addLast(new Handler(server));
+              }
+            });
+
+    Channel channel = b.bind(address).syncUninterruptibly().channel();
+
+    InetSocketAddress socketAddress = (InetSocketAddress) channel.localAddress();
+    return new FakeHttpServer(
+        URI.create("http://localhost:" + socketAddress.getPort()),
+        channel,
+        () -> {
+          bossGroup.shutdownGracefully();
+          workerGroup.shutdownGracefully();
+        });
+  }
+
+  interface HttpRequestHandler {
+    FullHttpResponse apply(HttpRequest req) throws Exception;
+  }
+
+  /**
+   * Based on
+   * https://github.com/netty/netty/blob/59aa6e635b9996cf21cd946e64353270679adc73/example/src/main/java/io/netty/example/http/helloworld/HttpHelloWorldServerHandler.java
+   */
+  private static final class Handler extends SimpleChannelInboundHandler<HttpRequest> {
+
+    private final HttpRequestHandler server;
+
+    private Handler(HttpRequestHandler server) {
+      this.server = server;
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+      ctx.flush();
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, HttpRequest req) throws Exception {
+      FullHttpResponse resp = server.apply(req);
+      HttpHeaders headers = resp.headers();
+      if (!headers.contains(CONTENT_LENGTH)) {
+        ByteBuf content = resp.content();
+        headers.setInt(CONTENT_LENGTH, content.readableBytes());
+      }
+      headers.set(CONNECTION, CLOSE);
+      ChannelFuture future = ctx.writeAndFlush(resp);
+      future.addListener(ChannelFutureListener.CLOSE);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+      cause.printStackTrace();
+      ctx.close();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
index ad3621e6..ac192a1f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java
@@ -241,12 +241,11 @@ public final class ITGrpcStorageImplUploadRetryTest {
         StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
 
     private static final ChecksummedData checksummedData =
-        TestUtils.getChecksummedData(ByteString.copyFrom(bytes), Hasher.enabled());
+        TestUtils.getChecksummedData(ByteString.copyFrom(bytes), Hasher.noop());
     private static final WriteObjectRequest req1 =
         WriteObjectRequest.newBuilder()
             .setUploadId(uploadId)
             .setChecksummedData(checksummedData)
-            .setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(checksummedData.getCrc32C()))
             .setFinishWrite(true)
             .build();
     private static final WriteObjectResponse resp1 =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
new file mode 100644
index 00000000..18704dc2
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java
@@ -0,0 +1,814 @@
+/*
+ * 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;
+
+import static com.google.cloud.storage.ByteSizeConstants._128KiBL;
+import static com.google.cloud.storage.ByteSizeConstants._256KiBL;
+import static com.google.cloud.storage.ByteSizeConstants._512KiBL;
+import static com.google.cloud.storage.ByteSizeConstants._768KiBL;
+import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_RANGE;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler;
+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.ParallelFriendly;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.netty.shaded.io.netty.buffer.ByteBuf;
+import io.grpc.netty.shaded.io.netty.buffer.Unpooled;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus;
+import java.math.BigInteger;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
+import org.checkerframework.checker.nullness.qual.Nullable;
+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)
+@SingleBackend(Backend.PROD)
+@ParallelFriendly
+public final class ITJsonResumableSessionPutTaskTest {
+  private static final GsonFactory gson = GsonFactory.getDefaultInstance();
+  private static final NetHttpTransport transport = new NetHttpTransport.Builder().build();
+  private static final HttpResponseStatus RESUME_INCOMPLETE =
+      HttpResponseStatus.valueOf(308, "Resume Incomplete");
+  private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE =
+      HttpResponseStatus.valueOf(503, "");
+  private HttpClientContext httpClientContext;
+
+  @Rule public final TemporaryFolder temp = new TemporaryFolder();
+
+  @Before
+  public void setUp() throws Exception {
+    httpClientContext =
+        HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson));
+  }
+
+  @Test
+  public void emptyObjectHappyPath() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          so.setName("object-name").setSize(BigInteger.ZERO);
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 0L), 0));
+
+      ResumableOperationResult<@Nullable StorageObject> operationResult = task.call();
+      StorageObject object = operationResult.getObject();
+      assertThat(object).isNotNull();
+      assertThat(operationResult.getPersistedSize()).isEqualTo(0L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.9</h4>
+   *
+   * Partial successful append to session
+   *
+   * <p>The client has sent N bytes, the server confirmed N bytes as committed. The client sends K
+   * bytes starting at offset N. The server responds with only N + L with 0 &lt;= L &lt; K bytes as
+   * committed.
+   */
+  @Test
+  public void scenario9() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          String contentRangeString = req.headers().get(CONTENT_RANGE);
+          HttpContentRange parse = HttpContentRange.parse(contentRangeString);
+          long endInclusive = ((HttpContentRange.HasRange<?>) parse).range().endOffsetInclusive();
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive - 1);
+          resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(503);
+      assertThat(confirmedBytes.get()).isEqualTo(-1L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.7</h4>
+   *
+   * GCS Acknowledges more bytes than were sent in the PUT
+   *
+   * <p>The client believes the server offset is N, it sends K bytes and the server responds that N
+   * + 2K bytes are now committed.
+   *
+   * <p>The client has detected data loss and should raise an error and prevent sending of more
+   * bytes.
+   */
+  @Test
+  public void scenario7() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          String contentRangeString = req.headers().get(CONTENT_RANGE);
+          HttpContentRange parse = HttpContentRange.parse(contentRangeString);
+          long endInclusive = ((HttpContentRange.HasRange<?>) parse).range().endOffsetInclusive();
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive + 1);
+          resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+      assertThat(confirmedBytes.get()).isEqualTo(-1L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.1</h4>
+   *
+   * Attempting to append to a session which has already been finalized should raise an error
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * resource = { name= obj, persisted_size = 524288 }
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 0, data = [0:262144]
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes 0-262143/*
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 200 OK
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 524288}
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario1() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          URI uri = URI.create(req.uri());
+          so.setName("object")
+              .setBucket("bucket")
+              .setGeneration(1L)
+              .setMetageneration(1L)
+              .setSize(BigInteger.valueOf(_512KiBL))
+              .setMetadata(ImmutableMap.of("upload_id", uri.toString()));
+
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler);
+        TmpFile tmpFile =
+            DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _256KiBL)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.of(tmpFile.getPath()),
+              HttpContentRange.of(ByteRangeSpec.explicit(0L, _256KiBL)));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("invalid");
+      assertThat(confirmedBytes.get()).isEqualTo(-1L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.2</h4>
+   *
+   * Attempting to finalize a session with fewer bytes than GCS acknowledges.
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * persisted_size = 524288
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 262144, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes *&#47;262144
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 308 Resume Incomplete
+   * Range: bytes=0-524287
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario2() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          ByteRangeSpec range = ByteRangeSpec.explicit(0L, _512KiBL);
+          resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(_256KiBL));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("invalid");
+      assertThat(confirmedBytes.get()).isEqualTo(-1L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.3</h4>
+   *
+   * Attempting to finalize a session with more bytes than GCS acknowledges.
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * persisted_size = 262144
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes *&#47;524288
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 308 Resume Incomplete
+   * Range: bytes=0-262143
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario3() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          ByteRangeSpec range = ByteRangeSpec.explicit(0L, _256KiBL);
+          resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(_512KiBL));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+      assertThat(confirmedBytes.get()).isEqualTo(-1L);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.4</h4>
+   *
+   * Attempting to finalize an already finalized session
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * resource = {name = obj1, size = 262114}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 262114, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes *&#47;262114
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          URI uri = URI.create(req.uri());
+          so.setName("object")
+              .setBucket("bucket")
+              .setGeneration(1L)
+              .setMetageneration(1L)
+              .setSize(BigInteger.valueOf(_256KiBL))
+              .setMetadata(ImmutableMap.of("upload_id", uri.toString()));
+
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(_256KiBL));
+
+      ResumableOperationResult<@Nullable StorageObject> operationResult = task.call();
+      StorageObject call = operationResult.getObject();
+      assertThat(call).isNotNull();
+      assertThat(call.getMetadata())
+          .containsEntry("upload_id", uploadUrl.substring(endpoint.toString().length()));
+      assertThat(operationResult.getPersistedSize()).isEqualTo(_256KiBL);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.4.1</h4>
+   *
+   * Attempting to finalize an already finalized session (ack &lt; expected)
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * resource = {name = obj1, size = 262114}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes *&#47;524288
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_1() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          URI uri = URI.create(req.uri());
+          so.setName("object")
+              .setBucket("bucket")
+              .setGeneration(1L)
+              .setMetageneration(1L)
+              .setSize(BigInteger.valueOf(_256KiBL))
+              .setMetadata(ImmutableMap.of("upload_id", uri.toString()));
+
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(_512KiBL));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+      assertThat(confirmedBytes.get()).isEqualTo(-1);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.4.2</h4>
+   *
+   * Attempting to finalize an already finalized session (ack > expected)
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * resource = {name = obj1, size = 262114}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes *&#47;131072
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_2() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          URI uri = URI.create(req.uri());
+          so.setName("object")
+              .setBucket("bucket")
+              .setGeneration(1L)
+              .setMetageneration(1L)
+              .setSize(BigInteger.valueOf(_256KiBL))
+              .setMetadata(ImmutableMap.of("upload_id", uri.toString()));
+
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.empty(),
+              HttpContentRange.of(_128KiBL));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+      assertThat(confirmedBytes.get()).isEqualTo(-1);
+    }
+  }
+
+  /**
+   *
+   *
+   * <h4>S.5</h4>
+   *
+   * Attempt to append to a resumable session with an offset higher than GCS expects
+   *
+   * <table>
+   *   <caption></caption>
+   *   <tr>
+   *     <td>server state</td>
+   *     <td><pre>
+   * persisted_size = 262144
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, data = [524288:786432]
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes 524288-786431&#47;*
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * 503
+   * Content-Type: text/plain; charset=utf-8
+   *
+   * Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s).
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario5() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          // error message from GCS circa 2023-02
+          ByteBuf buf =
+              Unpooled.wrappedBuffer(
+                  "Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s)."
+                      .getBytes(StandardCharsets.UTF_8));
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(
+                  req.protocolVersion(), APPEND_GREATER_THAN_CURRENT_SIZE, buf);
+          resp.headers().set(CONTENT_TYPE, "text/plain; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler);
+        TmpFile tmpFile =
+            DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _256KiBL)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableHttpContent.of(tmpFile.getPath()),
+              HttpContentRange.of(ByteRangeSpec.explicit(_512KiBL, _768KiBL)));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+      assertThat(confirmedBytes.get()).isEqualTo(-1);
+    }
+  }
+
+  @Test
+  public void jsonParseFailure() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          URI uri = URI.create(req.uri());
+          so.setName("object")
+              .setBucket("bucket")
+              .setGeneration(1L)
+              .setMetageneration(1L)
+              .setSize(BigInteger.ZERO)
+              .setMetadata(ImmutableMap.of("upload_id", uri.toString()));
+
+          byte[] bytes = gson.toByteArray(so);
+          ByteBuf buf = Unpooled.wrappedBuffer(bytes, 0, bytes.length / 2);
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext, uploadUrl, RewindableHttpContent.empty(), HttpContentRange.of(0));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      // the parse error happens while trying to read the success object, make sure we raise it as
+      // a client side retryable exception
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo(null);
+      // Finalization was successful, but we can't confirm the number of bytes due to the parse
+      // error
+      assertThat(confirmedBytes.get()).isEqualTo(-1);
+
+      ResultRetryAlgorithm<?> idempotentHandler =
+          StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler();
+      boolean shouldRetry = idempotentHandler.shouldRetry(se, null);
+      assertThat(shouldRetry).isTrue();
+    }
+  }
+
+  @Test
+  public void jsonDeserializationOnlyAttemptedWhenContentPresent() throws Exception {
+
+    HttpRequestHandler handler =
+        req -> {
+          DefaultFullHttpResponse resp = new DefaultFullHttpResponse(req.protocolVersion(), OK);
+          resp.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
+          resp.headers().set("x-goog-stored-content-length", "0");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext, uploadUrl, RewindableHttpContent.empty(), HttpContentRange.of(0));
+
+      ResumableOperationResult<@Nullable StorageObject> operationResult = task.call();
+      StorageObject call = operationResult.getObject();
+      assertThat(call).isNull();
+      assertThat(operationResult.getPersistedSize()).isEqualTo(0L);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java
new file mode 100644
index 00000000..07a04ed6
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java
@@ -0,0 +1,235 @@
+/*
+ * 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;
+
+import static com.google.cloud.storage.ByteSizeConstants._256KiBL;
+import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler;
+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.ParallelFriendly;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import io.grpc.netty.shaded.io.netty.buffer.ByteBuf;
+import io.grpc.netty.shaded.io.netty.buffer.Unpooled;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus;
+import java.math.BigInteger;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@SingleBackend(Backend.PROD)
+@ParallelFriendly
+public final class ITJsonResumableSessionQueryTaskTest {
+  private static final GsonFactory gson = GsonFactory.getDefaultInstance();
+  private static final NetHttpTransport transport = new NetHttpTransport.Builder().build();
+  private static final HttpResponseStatus RESUME_INCOMPLETE =
+      HttpResponseStatus.valueOf(308, "Resume Incomplete");
+  private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE =
+      HttpResponseStatus.valueOf(503, "");
+
+  private HttpClientContext httpClientContext;
+
+  @Before
+  public void setUp() throws Exception {
+    httpClientContext =
+        HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson));
+  }
+
+  @Test
+  public void successfulSession() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          so.setName("object-name").setSize(BigInteger.ZERO);
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      ResumableOperationResult<@Nullable StorageObject> result = task.call();
+      StorageObject object = result.getObject();
+      assertThat(object).isNotNull();
+      assertThat(result.getPersistedSize()).isEqualTo(0L);
+    }
+  }
+
+  @Test
+  public void successfulSession_noObject() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          DefaultFullHttpResponse response = new DefaultFullHttpResponse(req.protocolVersion(), OK);
+          response.headers().set("X-Goog-Stored-Content-Length", 0);
+          return response;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      ResumableOperationResult<@Nullable StorageObject> result = task.call();
+      StorageObject object = result.getObject();
+      assertThat(object).isNull();
+      assertThat(result.getPersistedSize()).isEqualTo(0L);
+    }
+  }
+
+  @Test
+  public void incompleteSession() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          DefaultFullHttpResponse response =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          response
+              .headers()
+              .set(
+                  HttpHeaderNames.RANGE,
+                  ByteRangeSpec.relativeLength(0L, _256KiBL).getHttpRangeHeader());
+          return response;
+        };
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      ResumableOperationResult<@Nullable StorageObject> result = task.call();
+      assertThat(result.getPersistedSize()).isEqualTo(_256KiBL);
+    }
+  }
+
+  /**
+   * This is a hard failure from the perspective of GCS as a range header is a required header to be
+   * included in the response to a query upload request.
+   */
+  @Test
+  public void incompleteSession_missingRangeHeader() throws Exception {
+    HttpRequestHandler handler =
+        req -> new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(503);
+      assertThat(se).hasMessageThat().contains("Range");
+    }
+  }
+
+  @Test
+  public void successfulSession_noJson_noStoredContentLength() throws Exception {
+    HttpRequestHandler handler = req -> new DefaultFullHttpResponse(req.protocolVersion(), OK);
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+    }
+  }
+
+  @Test
+  public void successfulSession_noSize() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          StorageObject so = new StorageObject();
+          so.setName("object-name");
+          ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so));
+
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), OK, buf);
+          resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+    }
+  }
+
+  @Test
+  public void query_badOffset() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          // error message from GCS circa 2023-02
+          ByteBuf buf =
+              Unpooled.wrappedBuffer(
+                  "Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s)."
+                      .getBytes(StandardCharsets.UTF_8));
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(
+                  req.protocolVersion(), APPEND_GREATER_THAN_CURRENT_SIZE, buf);
+          resp.headers().set(CONTENT_TYPE, "text/plain; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableSessionQueryTask task =
+          new JsonResumableSessionQueryTask(httpClientContext, uploadUrl);
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(0);
+      assertThat(se.getReason()).isEqualTo("dataLoss");
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java
new file mode 100644
index 00000000..a71f5cf4
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.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;
+
+import static com.google.cloud.storage.ByteSizeConstants._256KiBL;
+import static com.google.cloud.storage.ByteSizeConstants._512KiBL;
+import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_RANGE;
+import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.RANGE;
+
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.core.ApiClock;
+import com.google.api.core.NanoClock;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpRequest;
+import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public final class ITJsonResumableSessionTest {
+  private static final GsonFactory gson = GsonFactory.getDefaultInstance();
+  private static final NetHttpTransport transport = new NetHttpTransport.Builder().build();
+  private static final HttpResponseStatus RESUME_INCOMPLETE =
+      HttpResponseStatus.valueOf(308, "Resume Incomplete");
+  private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE =
+      HttpResponseStatus.valueOf(503, "");
+  private static final RetryingDependencies RETRYING_DEPENDENCIES =
+      new RetryingDependencies() {
+        @Override
+        public RetrySettings getRetrySettings() {
+          return RetrySettings.newBuilder().setMaxAttempts(3).build();
+        }
+
+        @Override
+        public ApiClock getClock() {
+          return NanoClock.getDefaultClock();
+        }
+      };
+  private static final ResultRetryAlgorithm<?> RETRY_ALGORITHM =
+      StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler();
+  private HttpClientContext httpClientContext;
+
+  @Rule public final TemporaryFolder temp = new TemporaryFolder();
+
+  @Before
+  public void setUp() throws Exception {
+    httpClientContext =
+        HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson));
+  }
+
+  @Test
+  public void rewindWillQueryStatusOnlyWhenDirty() throws Exception {
+    HttpContentRange range1 = HttpContentRange.of(ByteRangeSpec.explicit(0L, _512KiBL));
+    HttpContentRange range2 = HttpContentRange.query();
+    HttpContentRange range3 = HttpContentRange.of(ByteRangeSpec.explicit(_256KiBL, _512KiBL));
+
+    final List<HttpRequest> requests = Collections.synchronizedList(new ArrayList<>());
+    HttpRequestHandler handler =
+        req -> {
+          requests.add(req);
+          String contentRange = req.headers().get(CONTENT_RANGE);
+          System.out.println("contentRange = " + contentRange);
+          DefaultFullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
+          if (range1.getHeaderValue().equals(contentRange)) {
+            resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
+          } else if (range2.getHeaderValue().equals(contentRange)) {
+            resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
+          } else {
+            resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _512KiBL).getHttpRangeHeader());
+          }
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler);
+        TmpFile tmpFile =
+            DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _512KiBL)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      JsonResumableWrite resumableWrite = JsonResumableWrite.of(null, ImmutableMap.of(), uploadUrl);
+      JsonResumableSession session =
+          new JsonResumableSession(
+              httpClientContext, RETRYING_DEPENDENCIES, RETRY_ALGORITHM, resumableWrite);
+
+      ResumableOperationResult<@Nullable StorageObject> operationResult =
+          session.put(RewindableHttpContent.of(tmpFile.getPath()), range1);
+      StorageObject call = operationResult.getObject();
+      assertThat(call).isNull();
+      assertThat(operationResult.getPersistedSize()).isEqualTo(_512KiBL);
+    }
+
+    assertThat(requests).hasSize(3);
+    List<String> actual =
+        requests.stream().map(r -> r.headers().get(CONTENT_RANGE)).collect(Collectors.toList());
+
+    List<String> expected =
+        ImmutableList.of(range1.getHeaderValue(), range2.getHeaderValue(), range3.getHeaderValue());
+
+    assertThat(actual).isEqualTo(expected);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java
new file mode 100644
index 00000000..5f23a5aa
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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;
+
+import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isContinue;
+import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isOk;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.client.http.EmptyContent;
+import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpRequest;
+import com.google.api.client.http.HttpResponse;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.testing.http.MockHttpTransport;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.common.collect.ImmutableMap;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import org.junit.Test;
+
+public final class JsonResumableSessionFailureScenarioTest {
+  private static final GsonFactory gson = GsonFactory.getDefaultInstance();
+
+  @Test
+  public void isOk_200() {
+    assertThat(isOk(200)).isTrue();
+  }
+
+  @Test
+  public void isOk_201() {
+    assertThat(isOk(201)).isTrue();
+  }
+
+  @Test
+  public void isContinue_308() {
+    assertThat(isContinue(308)).isTrue();
+  }
+
+  @Test
+  public void toStorageException_ioExceptionDuringContentResolutionAddedAsSuppressed()
+      throws IOException {
+    HttpRequest req =
+        new MockHttpTransport()
+            .createRequestFactory()
+            .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent());
+    req.getHeaders().setContentLength(0L).setContentRange(HttpContentRange.of(0).getHeaderValue());
+
+    HttpResponse resp = req.execute();
+    resp.getHeaders().setContentType("text/plain; charset=utf-8").setContentLength(5L);
+
+    StorageException storageException =
+        JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException(
+            "uploadId",
+            resp,
+            new Cause(),
+            () -> {
+              throw new Kaboom();
+            });
+
+    assertThat(storageException.getCode()).isEqualTo(400);
+    assertThat(storageException).hasCauseThat().isInstanceOf(Cause.class);
+    assertThat(storageException.getSuppressed()).isNotEmpty();
+    assertThat(storageException.getSuppressed()[0]).isInstanceOf(StorageException.class);
+    assertThat(storageException.getSuppressed()[0]).hasCauseThat().isInstanceOf(Kaboom.class);
+  }
+
+  @Test
+  public void multilineResponseBodyIsProperlyPrefixed() throws Exception {
+    StorageObject so = new StorageObject();
+    so.setName("object-name")
+        .setSize(BigInteger.ZERO)
+        .setGeneration(1L)
+        .setMetageneration(2L)
+        .setMetadata(
+            ImmutableMap.of(
+                "k1", "v1",
+                "k2", "v2"));
+    final String json = gson.toPrettyString(so);
+
+    byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
+    HttpRequest req =
+        new MockHttpTransport()
+            .createRequestFactory()
+            .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent());
+    req.getHeaders().setContentLength(0L);
+
+    HttpResponse resp = req.execute();
+    resp.getHeaders()
+        .setContentType("application/json; charset=utf-8")
+        .setContentLength((long) bytes.length);
+
+    StorageException storageException =
+        JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+            "uploadId", resp, null, () -> json);
+
+    assertThat(storageException.getCode()).isEqualTo(0);
+    assertThat(storageException).hasMessageThat().contains("\t|<   \"generation\": \"1\",\n");
+  }
+
+  @Test
+  public void xGoogStoredHeadersIncludedIfPresent() throws IOException {
+    HttpRequest req =
+        new MockHttpTransport()
+            .createRequestFactory()
+            .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent());
+    req.getHeaders().setContentLength(0L);
+
+    HttpResponse resp = req.execute();
+    resp.getHeaders()
+        .set("X-Goog-Stored-Content-Length", "5")
+        .set("x-goog-stored-content-encoding", "identity")
+        .set("X-GOOG-STORED-SOMETHING", "blah")
+        .setContentLength(0L);
+
+    StorageException storageException =
+        JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+            "uploadId", resp, null, () -> null);
+
+    assertThat(storageException.getCode()).isEqualTo(0);
+    assertThat(storageException).hasMessageThat().contains("|< x-goog-stored-content-length: 5");
+    assertThat(storageException)
+        .hasMessageThat()
+        .contains("|< x-goog-stored-content-encoding: identity");
+    assertThat(storageException).hasMessageThat().contains("|< x-goog-stored-something: blah");
+  }
+
+  private static final class Cause extends RuntimeException {
+
+    private Cause() {
+      super("Cause");
+    }
+  }
+
+  private static final class Kaboom extends IOException {
+
+    private Kaboom() {
+      super("Kaboom!!!");
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java
new file mode 100644
index 00000000..5d235377
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java
@@ -0,0 +1,201 @@
+/*
+ * 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;
+
+import static com.google.cloud.storage.TestUtils.xxd;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.common.base.MoreObjects;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import net.jqwik.api.Arbitraries;
+import net.jqwik.api.Arbitrary;
+import net.jqwik.api.Combinators;
+import net.jqwik.api.ForAll;
+import net.jqwik.api.Property;
+import net.jqwik.api.Provide;
+import net.jqwik.api.RandomDistribution;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class RewindableHttpContentPropertyTest {
+
+  @Property
+  void path(@ForAll("PathScenario") PathScenario pathScenario) throws Exception {
+    try (PathScenario s = pathScenario) {
+      RewindableHttpContent content = RewindableHttpContent.of(s.getPath());
+      assertThrows(
+          IOException.class,
+          () -> {
+            try (ErroringOutputStream erroringOutputStream =
+                new ErroringOutputStream(s.getErrorAtOffset())) {
+              content.writeTo(erroringOutputStream);
+            }
+          });
+      content.rewindTo(s.getRewindOffset());
+
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      content.writeTo(baos);
+
+      String actual = xxd(baos.toByteArray());
+
+      assertThat(actual).isEqualTo(s.getExpectedXxd());
+    }
+  }
+
+  @Provide("PathScenario")
+  static Arbitrary<PathScenario> pathScenario() {
+    return Arbitraries.lazyOf(
+        () ->
+            Arbitraries.oneOf(
+                    bytes(1, 10),
+                    bytes(10, 100),
+                    bytes(100, 1_000),
+                    bytes(1_000, 10_000),
+                    bytes(10_000, 100_000),
+                    bytes(100_000, 1_000_000),
+                    bytes(1_000_000, 10_000_000))
+                .flatMap(
+                    bytes ->
+                        Combinators.combine(
+                                Arbitraries.integers().between(0, bytes.length - 1),
+                                Arbitraries.integers().between(0, bytes.length - 1),
+                                Arbitraries.just(bytes))
+                            .as(PathScenario::of)));
+  }
+
+  @NonNull
+  private static Arbitrary<byte[]> bytes(int minFileSize, int maxFileSize) {
+    return Arbitraries.integers()
+        .between(minFileSize, maxFileSize)
+        .withDistribution(RandomDistribution.uniform())
+        .map(DataGenerator.base64Characters()::genBytes);
+  }
+
+  private static final class PathScenario implements AutoCloseable {
+
+    private static final Path TMP_DIR = Paths.get(System.getProperty("java.io.tmpdir"));
+
+    private final int rewindOffset;
+    private final int errorAtOffset;
+    private final TmpFile tmpFile;
+    private final byte[] expectedBytes;
+    private final String expectedXxd;
+
+    private PathScenario(
+        int rewindOffset, int errorAtOffset, TmpFile tmpFile, byte[] expectedBytes) {
+      this.rewindOffset = rewindOffset;
+      this.errorAtOffset = errorAtOffset;
+      this.tmpFile = tmpFile;
+      this.expectedBytes = expectedBytes;
+      this.expectedXxd = xxd(expectedBytes);
+    }
+
+    public int getRewindOffset() {
+      return rewindOffset;
+    }
+
+    public int getErrorAtOffset() {
+      return errorAtOffset;
+    }
+
+    public Path getPath() {
+      return tmpFile.getPath();
+    }
+
+    public String getExpectedXxd() {
+      return expectedXxd;
+    }
+
+    public long getFullLength() throws IOException {
+      return Files.size(tmpFile.getPath());
+    }
+
+    @Override
+    public void close() throws IOException {
+      tmpFile.close();
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("expectedXxd", "\n" + expectedXxd)
+          .add("expectedBytes.length", expectedBytes.length)
+          .add("rewindOffset", rewindOffset)
+          .add("errorAtOffset", errorAtOffset)
+          .add("tmpFile", tmpFile)
+          .toString();
+    }
+
+    private static PathScenario of(int rewindOffset, int errorAtOffset, byte[] bytes) {
+      try {
+        TmpFile tmpFile1 = TmpFile.of(TMP_DIR, "PathScenario", ".bin");
+        try (SeekableByteChannel writer = tmpFile1.writer()) {
+          writer.write(ByteBuffer.wrap(bytes));
+        }
+        byte[] expectedBytes =
+            Arrays.copyOfRange(bytes, Math.min(rewindOffset, bytes.length), bytes.length);
+        return new PathScenario(rewindOffset, errorAtOffset, tmpFile1, expectedBytes);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
+
+  static final class ErroringOutputStream extends OutputStream {
+    private final long errorAt;
+    private long totalWritten;
+
+    ErroringOutputStream(long errorAt) {
+      this.errorAt = errorAt;
+      this.totalWritten = 0;
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+      if (totalWritten++ >= errorAt) {
+        throw new IOException("Reached errorAt limit");
+      }
+    }
+
+    @Override
+    public void write(byte[] b) throws IOException {
+      if (totalWritten + b.length >= errorAt) {
+        throw new IOException("Reached errorAt limit");
+      } else {
+        totalWritten += b.length;
+      }
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+      int diff = len - off;
+      if (totalWritten + diff >= errorAt) {
+        throw new IOException("Reached errorAt limit");
+      } else {
+        totalWritten += diff;
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
index ce750154..8395c6d0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
@@ -983,17 +983,6 @@ public class StorageImplMockitoTest {
     return blobInfo;
   }
 
-  @Test
-  public void testCreateFromFile() throws Exception {
-    byte[] dataToSend = {1, 2, 3, 4};
-    Path tempFile = Files.createTempFile("testCreateFrom", ".tmp");
-    Files.write(tempFile, dataToSend);
-
-    BlobInfo blobInfo = initializeUpload(dataToSend);
-    Blob blob = storage.createFrom(blobInfo, tempFile);
-    assertEquals(expectedUpdated, blob);
-  }
-
   @Test
   public void testCreateFromStream() throws Exception {
     byte[] dataToSend = {1, 2, 3, 4, 5};
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java
index 46d5643e..47ab7a82 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java
@@ -37,6 +37,10 @@ final class Functions {
       return (Ctx ctx, TestRetryConformance trc) -> f.apply(apply(ctx, trc), trc);
     }
 
+    default CtxFunction compose(CtxFunction f) {
+      return (Ctx ctx, TestRetryConformance trc) -> apply(f.apply(ctx, trc), trc);
+    }
+
     static CtxFunction identity() {
       return (ctx, c) -> ctx;
     }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
index 8aa55955..d42755ef 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java
@@ -60,6 +60,7 @@ import java.util.List;
 import java.util.Random;
 import java.util.Set;
 import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import org.junit.After;
@@ -209,7 +210,7 @@ public class ITRetryConformanceTest {
    * each defined scenario from google-cloud-conformance-tests and our defined {@link
    * RpcMethodMappings}.
    */
-  private static final class RetryTestCaseResolver {
+  static final class RetryTestCaseResolver {
     private static final String HEX_SHUFFLE_SEED_OVERRIDE =
         System.getProperty("HEX_SHUFFLE_SEED_OVERRIDE");
 
@@ -220,7 +221,7 @@ public class ITRetryConformanceTest {
     private final String host;
     private final String projectId;
 
-    RetryTestCaseResolver(
+    private RetryTestCaseResolver(
         String retryTestsJsonResourcePath,
         RpcMethodMappings mappings,
         BiPredicate<RpcMethod, TestRetryConformance> testAllowFilter,
@@ -383,8 +384,12 @@ public class ITRetryConformanceTest {
       return (m, c) -> set.contains(c.getMappingId());
     }
 
-    static BiPredicate<RpcMethod, TestRetryConformance> instructionsAre(String... instructions) {
-      return (m, trc) ->
+    static BiPredicate<RpcMethod, TestRetryConformance> lift(Predicate<TestRetryConformance> p) {
+      return (m, trc) -> p.test(trc);
+    }
+
+    static Predicate<TestRetryConformance> instructionsAre(String... instructions) {
+      return trc ->
           trc.getInstruction().getInstructionsList().equals(ImmutableList.copyOf(instructions));
     }
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
index 67b8159b..7a87f330 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java
@@ -20,10 +20,12 @@ import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSe
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.notificationSetup;
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.pubsubTopicSetup;
 import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.serviceAccount;
+import static com.google.cloud.storage.conformance.retry.ITRetryConformanceTest.RetryTestCaseResolver.instructionsAre;
 import static com.google.common.base.Predicates.and;
 import static com.google.common.base.Predicates.not;
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeFalse;
 
 import com.google.cloud.BaseServiceException;
 import com.google.cloud.Binding;
@@ -54,6 +56,7 @@ import com.google.cloud.storage.StorageRoles;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.Local;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup;
 import com.google.cloud.storage.conformance.retry.CtxFunctions.Rpc;
+import com.google.cloud.storage.conformance.retry.Functions.CtxFunction;
 import com.google.cloud.storage.conformance.retry.Functions.EConsumer;
 import com.google.cloud.storage.conformance.retry.RpcMethod.storage.bucket_acl;
 import com.google.cloud.storage.conformance.retry.RpcMethod.storage.buckets;
@@ -124,6 +127,11 @@ final class RpcMethodMappings {
 
   static final int _2MiB = 2 * 1024 * 1024;
   private static final ImmutableMap<String, String> MODIFY = ImmutableMap.of("a", "b");
+  private static final CtxFunction skipUntil2114Fixed =
+      temporarilySkipMapping(
+          "Skipped until https://github.com/googleapis/java-storage/issues/2114 is fixed",
+          instructionsAre("return-503-after-8192K", "return-408")
+              .or(instructionsAre("return-503-after-256K")));
   final Multimap<RpcMethod, RpcMethodMapping> funcMap;
 
   RpcMethodMappings() {
@@ -1526,7 +1534,10 @@ final class RpcMethodMappings {
         a.add(
             RpcMethodMapping.newBuilder(50, objects.insert)
                 .withApplicable(TestRetryConformance::isPreconditionsProvided)
-                .withSetup(defaultSetup.andThen(Local.blobInfoWithGenerationZero))
+                .withSetup(
+                    defaultSetup
+                        .andThen(Local.blobInfoWithGenerationZero)
+                        .compose(skipUntil2114Fixed))
                 .withTest(
                     (ctx, c) ->
                         ctx.map(
@@ -1541,7 +1552,10 @@ final class RpcMethodMappings {
         a.add(
             RpcMethodMapping.newBuilder(51, objects.insert)
                 .withApplicable(TestRetryConformance::isPreconditionsProvided)
-                .withSetup(defaultSetup.andThen(Local.blobInfoWithGenerationZero))
+                .withSetup(
+                    defaultSetup
+                        .andThen(Local.blobInfoWithGenerationZero)
+                        .compose(skipUntil2114Fixed))
                 .withTest(
                     (ctx, c) ->
                         ctx.map(
@@ -2099,4 +2113,12 @@ final class RpcMethodMappings {
   private static Predicate<TestRetryConformance> methodGroupIs(String s) {
     return (c) -> s.equals(c.getMethod().getGroup());
   }
+
+  private static CtxFunction temporarilySkipMapping(
+      String message, java.util.function.Predicate<TestRetryConformance> p) {
+    return (ctx, trc) -> {
+      assumeFalse(message, p.test(trc));
+      return ctx;
+    };
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
index 2bcb78e6..d8e026cb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
@@ -28,6 +28,7 @@ import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.TmpFile;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.ITObjectChecksumSupportTest.ChecksummedTestContentProvider;
 import com.google.cloud.storage.it.runner.StorageITRunner;
@@ -42,18 +43,24 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.io.ByteStreams;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
 @CrossRun(
-    transports = {Transport.HTTP, Transport.GRPC},
+    transports = {Transport.HTTP /*, Transport.GRPC*/},
     backends = Backend.PROD)
 @Parameterized(ChecksummedTestContentProvider.class)
 public final class ITObjectChecksumSupportTest {
 
+  private static final Path tmpDir = Paths.get(System.getProperty("java.io.tmpdir"));
+
   @Inject public Generator generator;
 
   @Inject public Storage storage;
@@ -116,6 +123,50 @@ public final class ITObjectChecksumSupportTest {
     assertThat(blob.getCrc32c()).isEqualTo(content.getCrc32cBase64());
   }
 
+  @Test
+  public void testCrc32cValidated_createFrom_path_expectFailure() throws IOException {
+    String blobName = generator.randomObjectName();
+    BlobId blobId = BlobId.of(bucket.getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
+
+    try (TmpFile tmpFile = TmpFile.of(tmpDir, "prefix", "bin")) {
+      try (SeekableByteChannel writer = tmpFile.writer()) {
+        writer.write(ByteBuffer.wrap(content.concat('x')));
+      }
+      StorageException expected =
+          assertThrows(
+              StorageException.class,
+              () ->
+                  storage.createFrom(
+                      blobInfo,
+                      tmpFile.getPath(),
+                      BlobWriteOption.doesNotExist(),
+                      BlobWriteOption.crc32cMatch()));
+      assertThat(expected.getCode()).isEqualTo(400);
+    }
+  }
+
+  @Test
+  public void testCrc32cValidated_createFrom_path_expectSuccess() throws IOException {
+    String blobName = generator.randomObjectName();
+    BlobId blobId = BlobId.of(bucket.getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
+
+    try (TmpFile tmpFile = TmpFile.of(tmpDir, "prefix", "bin")) {
+      try (SeekableByteChannel writer = tmpFile.writer()) {
+        writer.write(ByteBuffer.wrap(content.getBytes()));
+      }
+
+      Blob blob =
+          storage.createFrom(
+              blobInfo,
+              tmpFile.getPath(),
+              BlobWriteOption.doesNotExist(),
+              BlobWriteOption.crc32cMatch());
+      assertThat(blob.getCrc32c()).isEqualTo(content.getCrc32cBase64());
+    }
+  }
+
   @Test
   public void testCrc32cValidated_writer_expectFailure() {
     String blobName = generator.randomObjectName();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
new file mode 100644
index 00000000..7416109b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
@@ -0,0 +1,468 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.TestUtils;
+import com.google.cloud.storage.TmpFile;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.transfermanager.DownloadJob;
+import com.google.cloud.storage.transfermanager.DownloadResult;
+import com.google.cloud.storage.transfermanager.ParallelDownloadConfig;
+import com.google.cloud.storage.transfermanager.ParallelUploadConfig;
+import com.google.cloud.storage.transfermanager.TransferManager;
+import com.google.cloud.storage.transfermanager.TransferManagerConfig;
+import com.google.cloud.storage.transfermanager.TransferManagerConfigTestingInstances;
+import com.google.cloud.storage.transfermanager.TransferStatus;
+import com.google.cloud.storage.transfermanager.UploadJob;
+import com.google.cloud.storage.transfermanager.UploadResult;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    transports = {Transport.HTTP},
+    backends = {Backend.PROD})
+public class ITTransferManagerTest {
+
+  private static final Comparator<BlobInfo> comp =
+      Comparator.comparing(info -> info.getBlobId().getName());
+  private static final Comparator<DownloadResult> comp2 =
+      Comparator.comparing(DownloadResult::getInput, comp);
+
+  @Inject public Storage storage;
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  @Rule public final TemporaryFolder tmpDir = new TemporaryFolder();
+
+  private Path baseDir;
+  private static final int objectContentSize = 64;
+  private List<BlobInfo> blobs = new ArrayList<>();
+
+  @Before
+  public void setUp() throws Exception {
+    baseDir = tmpDir.getRoot().toPath();
+    BlobInfo blobInfo1 =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    BlobInfo blobInfo2 =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    BlobInfo blobInfoChunking =
+        BlobInfo.newBuilder(
+                BlobId.of(bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+            .build();
+    Collections.addAll(blobs, blobInfo1, blobInfo2);
+    ByteBuffer content = DataGenerator.base64Characters().genByteBuffer(108);
+    for (BlobInfo blob : blobs) {
+      try (WriteChannel writeChannel = storage.writer(blob)) {
+        writeChannel.write(content);
+      }
+    }
+    long size = 2L * 1024 * 1024;
+    size = size + 100L;
+    ByteBuffer chunkedContent = DataGenerator.base64Characters().genByteBuffer(size);
+    try (WriteChannel writeChannel = storage.writer(blobInfoChunking)) {
+      writeChannel.write(chunkedContent);
+    }
+    blobs.add(blobInfoChunking);
+  }
+
+  @Test
+  public void uploadFiles() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile1 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile2 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      List<Path> files =
+          ImmutableList.of(tmpFile.getPath(), tmpFile1.getPath(), tmpFile2.getPath());
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(3);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  public void uploadFilesWithOpts() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile1 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile2 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      List<Path> files =
+          ImmutableList.of(tmpFile.getPath(), tmpFile1.getPath(), tmpFile2.getPath());
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setWriteOptsPerRequest(Collections.singletonList(BlobWriteOption.doesNotExist()))
+              .build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(3);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  public void uploadFilesOneFailure() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile1 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
+        TmpFile tmpFile2 = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      List<Path> files =
+          ImmutableList.of(
+              tmpFile.getPath(),
+              tmpFile1.getPath(),
+              tmpFile2.getPath(),
+              Paths.get("this-file-does-not-exist.txt"));
+      String bucketName = bucket.getName();
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(4);
+      assertThat(
+              uploadResults.stream()
+                  .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_FINISH)
+                  .collect(Collectors.toList()))
+          .hasSize(1);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
+  @Test
+  public void uploadNonexistentBucket() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName() + "-does-not-exist";
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      List<Path> files = ImmutableList.of(tmpFile.getPath());
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
+      assertThat(uploadResults.get(0).getException()).isInstanceOf(StorageException.class);
+    }
+  }
+
+  @Test
+  public void uploadNonexistentFile() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService()) {
+      List<Path> files = ImmutableList.of(Paths.get("this-file-does-not-exist.txt"));
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
+      assertThat(uploadResults.get(0).getException()).isInstanceOf(NoSuchFileException.class);
+    }
+  }
+
+  @Test
+  public void uploadFailsSkipIfExists() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucketName).setSkipIfExists(true).build();
+      UploadJob jobInitUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+      UploadJob failedSecondUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
+      assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+    }
+  }
+
+  @Test
+  public void uploadSkipIfExistsGenerationOverride() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
+    String bucketName = bucket.getName();
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setSkipIfExists(true)
+              .setWriteOptsPerRequest(ImmutableList.of(BlobWriteOption.generationMatch(5L)))
+              .build();
+      assertThat(parallelUploadConfig.getWriteOptsPerRequest()).hasSize(1);
+      UploadJob jobInitUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+      UploadJob failedSecondUpload =
+          transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
+      assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+    }
+  }
+
+  @Test
+  public void downloadBlobs() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      try {
+        assertThat(downloadResults).hasSize(3);
+      } finally {
+        cleanUpFiles(downloadResults);
+      }
+    }
+  }
+
+  @Test
+  public void downloadBlobsAllowChunked() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      assertThat(downloadResults).hasSize(3);
+
+      List<String> expectedContents =
+          blobs.stream()
+              .sorted(comp)
+              .map(BlobInfo::getBlobId)
+              .map(storage::readAllBytes)
+              .map(TestUtils::xxd)
+              .collect(Collectors.toList());
+
+      List<String> actualContents =
+          downloadResults.stream()
+              .sorted(comp2)
+              .map(DownloadResult::getOutputDestination)
+              .map(ITTransferManagerTest::readAllPathBytes)
+              .map(TestUtils::xxd)
+              .collect(Collectors.toList());
+
+      try {
+        assertThat(actualContents).isEqualTo(expectedContents);
+      } finally {
+        cleanUpFiles(downloadResults);
+      }
+    }
+  }
+
+  @Test
+  public void downloadNonexistentBucket() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName() + "-does-not-exist";
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsChunkedFail() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName() + "-does-not-exist";
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      assertThat(downloadResults).hasSize(3);
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsPreconditionFailure() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowDivideAndConquer(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    try (TransferManager transferManager = config.getService()) {
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucket.getName())
+              .setDownloadDirectory(baseDir)
+              .setOptionsPerRequest(ImmutableList.of(BlobSourceOption.generationMatch(-1)))
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      assertThat(downloadResults).hasSize(3);
+      List<DownloadResult> failedToStart =
+          downloadResults.stream()
+              .filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
+              .collect(Collectors.toList());
+      assertThat(failedToStart).hasSize(3);
+    }
+  }
+
+  @Test
+  public void downloadBlobsOneFailure() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .build();
+      List<BlobInfo> downloadBlobs = blobs;
+      BlobInfo nonexistentBlob =
+          BlobInfo.newBuilder(
+                  BlobId.of(
+                      bucket.getName(), String.format("%s/src", generator.randomObjectName())))
+              .build();
+      downloadBlobs.add(nonexistentBlob);
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      try {
+        assertThat(downloadResults).hasSize(4);
+        assertThat(
+                downloadResults.stream()
+                    .filter(res -> res.getStatus() == TransferStatus.FAILED_TO_START)
+                    .collect(Collectors.toList()))
+            .hasSize(1);
+      } finally {
+        cleanUpFiles(
+            downloadResults.stream()
+                .filter(res -> res.getStatus() == TransferStatus.SUCCESS)
+                .collect(Collectors.toList()));
+      }
+    }
+  }
+
+  private void cleanUpFiles(List<DownloadResult> results) throws IOException {
+    // Cleanup downloaded blobs and the parent directory
+    for (DownloadResult res : results) {
+      Files.delete(res.getOutputDestination());
+      Files.delete(res.getOutputDestination().getParent());
+    }
+  }
+
+  private static byte[] readAllPathBytes(Path path) {
+    try {
+      return Files.readAllBytes(path);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
new file mode 100644
index 00000000..1c5e67b9
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+import com.google.cloud.storage.StorageOptions;
+
+public final class TransferManagerConfigTestingInstances {
+  private TransferManagerConfigTestingInstances() {}
+
+  public static TransferManagerConfig defaults() {
+    return defaults(StorageOptions.newBuilder().build());
+  }
+
+  public static TransferManagerConfig defaults(StorageOptions options) {
+    return TransferManagerConfig.newBuilder()
+        .setAllowDivideAndConquer(false)
+        .setMaxWorkers(1)
+        .setPerWorkerBufferSize(512 * 1024)
+        .setQos(DefaultQos.of(2 * 1024 * 1024))
+        .setStorageOptions(options)
+        .build();
+  }
+}