Repository: googleapis/java-storage
Pull request #2116 (2023-07-13 19:07:28)

test(deps): update dependency com.google.api.grpc:proto-google-cloud-kms-v1 to v0.115.0
[![Mend Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com)

This PR contains the following updates:

| Package | Change | Age | Adoption | Passing | Confidence |
|---|---|---|---|---|---|
| [com.google.api.grpc:proto-google-cloud-kms-v1](https://togithub.com/googleapis/google-cloud-java) | `0.114.0` -> `0.115.0` | [![age](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.115.0/age-slim)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.115.0/adoption-slim)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.115.0/compatibility-slim/0.114.0)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.115.0/confidence-slim/0.114.0)](https://docs.renovatebot.com/merge-confidence/) |

---

### Release Notes

<details>
<summary>googleapis/google-cloud-java (com.google.api.grpc:proto-google-cloud-kms-v1)</summary>

### [`v0.115.0`](https://togithub.com/googleapis/google-cloud-java/releases/tag/v0.115.0)

[Compare Source](https://togithub.com/googleapis/google-cloud-java/compare/v0.114.0...v0.115.0)

10-09-2019 15:50 PDT

##### Bigtable

-   Updated javaDoc for callable being wrapper around read & write ([#&#8203;6431](https://togithub.com/googleapis/google-cloud-java/pull/6431))
-   Updated JavaDoc to mention connection pooling over ManagedChannel ([#&#8203;6328](https://togithub.com/googleapis/google-cloud-java/pull/6328))
-   Overload MutationApi#setCell to accept long values ([#&#8203;6432](https://togithub.com/googleapis/google-cloud-java/pull/6432))
-   Updated Javadoc of models/CreateInstanceRequest.java ([#&#8203;6201](https://togithub.com/googleapis/google-cloud-java/pull/6201))
-   Re-enable the large rows test now that it's fixed ([#&#8203;6459](https://togithub.com/googleapis/google-cloud-java/pull/6459))

##### Dataproc

-   Regenerate dataproc client - autoscaling configuration ([#&#8203;6454](https://togithub.com/googleapis/google-cloud-java/pull/6454))
-   Regenerate dataproc client ([#&#8203;6403](https://togithub.com/googleapis/google-cloud-java/pull/6403))
-   Update dependency com.google.code.gson:gson to v2.8.6 ([#&#8203;6440](https://togithub.com/googleapis/google-cloud-java/pull/6440))

##### Datastore

-   fix a couple of warnings ([#&#8203;6458](https://togithub.com/googleapis/google-cloud-java/pull/6458))
-   Datastore : add GqlQuery#setNullBinding method ([#&#8203;6360](https://togithub.com/googleapis/google-cloud-java/pull/6360))

##### Grafeas

-   Grafeas: reclassify GrafeasInteropTest as an integration test ([#&#8203;6461](https://togithub.com/googleapis/google-cloud-java/pull/6461))

##### Monitoring

-   Refresh VPCSC tests for monitoring V3 ([#&#8203;6472](https://togithub.com/googleapis/google-cloud-java/pull/6472))

##### OS Login

-   Regenerate os-login client ([#&#8203;6465](https://togithub.com/googleapis/google-cloud-java/pull/6465))

##### Redis

-   Regenerate redis client - update retry settings ([#&#8203;6455](https://togithub.com/googleapis/google-cloud-java/pull/6455))
-   Redis: Added system test  ([#&#8203;6375](https://togithub.com/googleapis/google-cloud-java/pull/6375))

##### Spanner

-   Unflake InstanceAdminGaxTest ([#&#8203;6470](https://togithub.com/googleapis/google-cloud-java/pull/6470))
-   add tests for translating to SessionNotFoundException ([#&#8203;6438](https://togithub.com/googleapis/google-cloud-java/pull/6438))

##### Talent

-   Regenerate talent client - proto docs ([#&#8203;6466](https://togithub.com/googleapis/google-cloud-java/pull/6466))

##### Tasks

-   Regenerate tasks client ([#&#8203;6424](https://togithub.com/googleapis/google-cloud-java/pull/6424))

##### Webrisk

-   Regenerate webrisk client ([#&#8203;6467](https://togithub.com/googleapis/google-cloud-java/pull/6467))

##### Dependencies

-   Update dependency com.google.auth:google-auth-library-bom to v0.18.0 ([#&#8203;6473](https://togithub.com/googleapis/google-cloud-java/pull/6473))
-   Update dependency com.google.apis:google-api-services-dns to v1-rev20190923-1.30.3 ([#&#8203;6464](https://togithub.com/googleapis/google-cloud-java/pull/6464))
-   Update dependency io.netty:netty-tcnative-boringssl-static to v2.0.26.Final ([#&#8203;6315](https://togithub.com/googleapis/google-cloud-java/pull/6315))
-   Update dependency com.google.api.grpc:grpc-google-common-protos to v1.17.0 ([#&#8203;6441](https://togithub.com/googleapis/google-cloud-java/pull/6441))
-   Update dependency com.google.api.grpc:proto-google-common-protos to v1.17.0 ([#&#8203;6442](https://togithub.com/googleapis/google-cloud-java/pull/6442))
-   Update google-cloud-core to 1.91.2 ([#&#8203;6448](https://togithub.com/googleapis/google-cloud-java/pull/6448))

##### Documentation

-   Recommend libraries-bom for troubleshooting ([#&#8203;6463](https://togithub.com/googleapis/google-cloud-java/pull/6463))
-   Be more direct about setting up ADC ([#&#8203;6457](https://togithub.com/googleapis/google-cloud-java/pull/6457))

##### Internal / Testing Changes

-   Extract gcloud-maven-plugin, bigtable-emulator ([#&#8203;6452](https://togithub.com/googleapis/google-cloud-java/pull/6452))
-   Ignore Eclipse metadata ([#&#8203;6450](https://togithub.com/googleapis/google-cloud-java/pull/6450))
-   Prepare asset client for extraction ([#&#8203;6436](https://togithub.com/googleapis/google-cloud-java/pull/6436))
-   Bump next snapshot ([#&#8203;6434](https://togithub.com/googleapis/google-cloud-java/pull/6434))
-   Fix translate release skipping ([#&#8203;6433](https://togithub.com/googleapis/google-cloud-java/pull/6433))

</details>

---

### Configuration

📅 **Schedule**: Branch creation - At any time (no schedule defined), Automerge - At any time (no schedule defined).

🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

🔕 **Ignore**: Close this PR and you won't be reminded about this update again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [Mend Renovate](https://www.mend.io/free-developer-tools/renovate/). View repository job log [here](https://developer.mend.io/github/googleapis/java-storage).
<!--renovate-debug:eyJjcmVhdGVkSW5WZXIiOiIzNi41LjMiLCJ1cGRhdGVkSW5WZXIiOiIzNi41LjMiLCJ0YXJnZXRCcmFuY2giOiJtYWluIn0=-->



Patch:
diff --git a/README.md b/README.md
index e40d26ae..1edd542d 100644
--- a/README.md
+++ b/README.md
@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-storage'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.23.0'
+implementation 'com.google.cloud:google-cloud-storage:2.24.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.23.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.24.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -428,7 +428,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.23.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.24.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index ccfc4154..d501ed01 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -173,7 +173,7 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.114.0</version>
+      <version>0.115.0</version>
       <scope>test</scope>
     </dependency>
     <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/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


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/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();