Repository: googleapis/java-storage
Pull request #2580 (2024-06-07 01:21:54)

chore(deps): update dependency com.google.cloud:google-cloud-storage to v2.40.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.cloud:google-cloud-storage](https://togithub.com/googleapis/java-storage) | `2.39.0` -> `2.40.0` | [![age](https://developer.mend.io/api/mc/badges/age/maven/com.google.cloud:google-cloud-storage/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://developer.mend.io/api/mc/badges/adoption/maven/com.google.cloud:google-cloud-storage/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://developer.mend.io/api/mc/badges/compatibility/maven/com.google.cloud:google-cloud-storage/2.39.0/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://developer.mend.io/api/mc/badges/confidence/maven/com.google.cloud:google-cloud-storage/2.39.0/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) |

---

### Release Notes

<details>
<summary>googleapis/java-storage (com.google.cloud:google-cloud-storage)</summary>

### [`v2.40.0`](https://togithub.com/googleapis/java-storage/blob/HEAD/CHANGELOG.md#2400-2024-06-06)

##### Features

-   Promote google-cloud-storage-control to GA ([#&#8203;2575](https://togithub.com/googleapis/java-storage/issues/2575)) ([129f188](https://togithub.com/googleapis/java-storage/commit/129f188a9877b3bd71db3b00ba3d8d6e4095fb67))

##### Bug Fixes

-   Reduce Java 21 Virtual Thread Pinning in IO operations ([#&#8203;2553](https://togithub.com/googleapis/java-storage/issues/2553)) ([498fd0b](https://togithub.com/googleapis/java-storage/commit/498fd0b391c9d1d01c453fe31abea7890c0b498f))

##### Dependencies

-   Update dependency com.google.cloud:sdk-platform-java-config to v3.31.0 ([#&#8203;2571](https://togithub.com/googleapis/java-storage/issues/2571)) ([67ce3d6](https://togithub.com/googleapis/java-storage/commit/67ce3d6505e1e48f496ab52884f5f5c2852aa445))
-   Update dependency net.jqwik:jqwik to v1.8.5 ([#&#8203;2563](https://togithub.com/googleapis/java-storage/issues/2563)) ([88f7d86](https://togithub.com/googleapis/java-storage/commit/88f7d86d960aa6986cf41165d80d03a551502ec2))

##### Documentation

-   Managed Folders samples ([#&#8203;2562](https://togithub.com/googleapis/java-storage/issues/2562)) ([5ffc1f2](https://togithub.com/googleapis/java-storage/commit/5ffc1f2fa75ac2e1c0022301e02b6ea6627faf03))
-   Update javadoc for createFrom ([#&#8203;2522](https://togithub.com/googleapis/java-storage/issues/2522)) ([dc31e95](https://togithub.com/googleapis/java-storage/commit/dc31e95158e66d0eb4c85122edd29431a610a204))

</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:eyJjcmVhdGVkSW5WZXIiOiIzNy4zOTMuMCIsInVwZGF0ZWRJblZlciI6IjM3LjQxMC4xIiwidGFyZ2V0QnJhbmNoIjoibWFpbiIsImxhYmVscyI6W119-->


dpebot: /gcbrun
sydney-munro: /gcbrun

Patch:
diff --git a/.readme-partials.yaml b/.readme-partials.yaml
index a04ad1a0..cb154392 100644
--- a/.readme-partials.yaml
+++ b/.readme-partials.yaml
@@ -32,7 +32,7 @@ custom_content: |
   <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
   </dependency>
   ```
   
@@ -44,7 +44,7 @@ custom_content: |
   If you are using Gradle without BOM, add this to your dependencies:
   
   ```Groovy
-  implementation 'com.google.cloud:google-cloud-storage-control:2.40.0' <!-- {x-version-update:google-cloud-storage-control:current} -->
+  implementation 'com.google.cloud:google-cloud-storage-control:2.40.1-SNAPSHOT' <!-- {x-version-update:google-cloud-storage-control:current} -->
   ```
   
   #### Creating an authorized service object
diff --git a/README.md b/README.md
index b0deb965..f1ab2bff 100644
--- a/README.md
+++ b/README.md
@@ -46,12 +46,12 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.39.0</version>
+  <version>2.40.0</version>
 </dependency>
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-control</artifactId>
-  <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
 </dependency>
 
 ```
@@ -59,20 +59,20 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.40.0')
+implementation platform('com.google.cloud:libraries-bom:26.41.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.39.0'
+implementation 'com.google.cloud:google-cloud-storage:2.40.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.39.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.40.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -141,7 +141,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-control</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
 </dependency>
 ```
 
@@ -153,7 +153,7 @@ implementation 'com.google.cloud:google-cloud-storage-control'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage-control:2.40.0' <!-- {x-version-update:google-cloud-storage-control:current} -->
+implementation 'com.google.cloud:google-cloud-storage-control:2.40.1-SNAPSHOT' <!-- {x-version-update:google-cloud-storage-control:current} -->
 ```
 
 #### Creating an authorized service object
@@ -343,6 +343,12 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-storage/tree/
 | Set Public Access Prevention Enforced | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionEnforced.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionEnforced.java) |
 | Set Public Access Prevention Inherited | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionInherited.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionInherited.java) |
 | Set Retention Policy | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java) |
+| Create Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java) |
+| Create Hierarchical Namespace Bucket | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java) |
+| Delete Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java) |
+| Get Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java) |
+| List Folders | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java) |
+| Rename Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java) |
 | Activate Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/ActivateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/ActivateHmacKey.java) |
 | Create Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/CreateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/CreateHmacKey.java) |
 | Deactivate Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/DeactivateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/DeactivateHmacKey.java) |
@@ -498,7 +504,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.39.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.40.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/gapic-google-cloud-storage-v2/pom.xml b/gapic-google-cloud-storage-v2/pom.xml
index 32201798..39b6839e 100644
--- a/gapic-google-cloud-storage-v2/pom.xml
+++ b/gapic-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>gapic-google-cloud-storage-v2</artifactId>
-  <version>2.40.0-alpha</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+  <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
   <name>gapic-google-cloud-storage-v2</name>
   <description>GRPC library for gapic-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/google-cloud-storage-bom/pom.xml b/google-cloud-storage-bom/pom.xml
index c5250748..0af6b000 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -19,7 +19,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-bom</artifactId>
-  <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>pom</packaging>
   <parent>
     <groupId>com.google.cloud</groupId>
@@ -69,37 +69,37 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:gapic-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
     </dependencies>
   </dependencyManagement>
diff --git a/google-cloud-storage-control/pom.xml b/google-cloud-storage-control/pom.xml
index a9b22062..cf5c06b5 100644
--- a/google-cloud-storage-control/pom.xml
+++ b/google-cloud-storage-control/pom.xml
@@ -5,13 +5,13 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-control</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
     <name>google-cloud-storage-control</name>
     <description>GRPC library for google-cloud-storage-control</description>
     <parent>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-parent</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </parent>
 
     <dependencies>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index b9d33513..819ed4e3 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -2,7 +2,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>jar</packaging>
   <name>Google Cloud Storage</name>
   <url>https://github.com/googleapis/java-storage</url>
@@ -12,7 +12,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java
index dab7b247..0f5a378f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java
@@ -52,7 +52,7 @@ final class BidiResumableWrite implements BidiWriteObjectRequestBuilderFactory {
 
   @Override
   public BidiWriteObjectRequest.Builder newBuilder() {
-    return writeRequest.toBuilder();
+    return writeRequest.toBuilder().clearWriteObjectSpec();
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java
index 09e4177c..a458c079 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java
@@ -16,8 +16,6 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.StorageV2ProtoUtils.fmtProto;
-
 import com.google.cloud.storage.BidiWriteCtx.BidiWriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.storage.v2.BidiWriteObjectRequest;
@@ -84,36 +82,5 @@ final class BidiWriteCtx<RequestFactoryT extends BidiWriteObjectRequestBuilderFa
 
     @Nullable
     String bucketName();
-
-    static BidiSimpleWriteObjectRequestBuilderFactory simple(BidiWriteObjectRequest req) {
-      return new BidiSimpleWriteObjectRequestBuilderFactory(req);
-    }
-  }
-
-  static final class BidiSimpleWriteObjectRequestBuilderFactory
-      implements BidiWriteObjectRequestBuilderFactory {
-    private final BidiWriteObjectRequest req;
-
-    private BidiSimpleWriteObjectRequestBuilderFactory(BidiWriteObjectRequest req) {
-      this.req = req;
-    }
-
-    @Override
-    public BidiWriteObjectRequest.Builder newBuilder() {
-      return req.toBuilder();
-    }
-
-    @Override
-    public @Nullable String bucketName() {
-      if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
-        return req.getWriteObjectSpec().getResource().getBucket();
-      }
-      return null;
-    }
-
-    @Override
-    public String toString() {
-      return "SimpleBidiWriteObjectRequestBuilderFactory{" + "req=" + fmtProto(req) + '}';
-    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index 4b6c34ef..5a7ef181 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -159,7 +159,10 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
                           grpc.startResumableWrite(grpcCallContext, req);
                       return ResumableMedia.gapic()
                           .write()
-                          .byteChannel(grpc.storageClient.writeObjectCallable())
+                          .byteChannel(
+                              grpc.storageClient
+                                  .writeObjectCallable()
+                                  .withDefaultCallContext(grpcCallContext))
                           .setHasher(Hasher.noop())
                           .setByteStringStrategy(ByteStringStrategy.copy())
                           .resumable()
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
index 19aba735..15278616 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
@@ -21,23 +21,28 @@ import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.BidiStreamingCallable;
+import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BidiWriteObjectResponse;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
+import io.grpc.Status;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 import java.util.function.Supplier;
@@ -47,18 +52,19 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
   private final BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write;
   private final RetryingDependencies deps;
   private final ResultRetryAlgorithm<?> alg;
-  private final String bucketName;
-  private final Supplier<GrpcCallContext> baseContextSupplier;
   private final SettableApiFuture<BidiWriteObjectResponse> resultFuture;
   private final ChunkSegmenter chunkSegmenter;
 
   private final BidiWriteCtx<BidiResumableWrite> writeCtx;
+  private final GrpcCallContext context;
   private final BidiObserver responseObserver;
 
   private volatile ApiStreamObserver<BidiWriteObjectRequest> stream;
   private boolean open = true;
   private boolean first = true;
   private boolean finished = false;
+  private volatile BidiWriteObjectRequest lastWrittenRequest;
+  private volatile RewindableContent currentContent;
 
   GapicBidiUnbufferedWritableByteChannel(
       BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write,
@@ -66,18 +72,18 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       ResultRetryAlgorithm<?> alg,
       SettableApiFuture<BidiWriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
-      BidiResumableWrite requestFactory,
+      BidiWriteCtx<BidiResumableWrite> writeCtx,
       Supplier<GrpcCallContext> baseContextSupplier) {
     this.write = write;
     this.deps = deps;
     this.alg = alg;
-    this.baseContextSupplier = baseContextSupplier;
-    this.bucketName = requestFactory.bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
-    this.writeCtx = new BidiWriteCtx<>(requestFactory);
+    this.writeCtx = writeCtx;
     this.responseObserver = new BidiObserver();
+    String bucketName = writeCtx.getRequestFactory().bucketName();
+    this.context = contextWithBucketName(bucketName, baseContextSupplier.get());
   }
 
   @Override
@@ -102,22 +108,22 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     if (!open) {
       return;
     }
-    ApiStreamObserver<BidiWriteObjectRequest> openedStream = openedStream();
-    if (!finished) {
-      BidiWriteObjectRequest message = finishMessage();
-      try {
-        openedStream.onNext(message);
-        finished = true;
-        openedStream.onCompleted();
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
+    try {
+      if (!finished) {
+        BidiWriteObjectRequest message = finishMessage();
+        lastWrittenRequest = message;
+        flush(Collections.singletonList(message));
+      } else {
+        if (stream != null) {
+          stream.onCompleted();
+          responseObserver.await();
+        }
       }
-    } else {
-      openedStream.onCompleted();
+    } finally {
+      open = false;
+      stream = null;
+      lastWrittenRequest = null;
     }
-    responseObserver.await();
-    open = false;
   }
 
   @VisibleForTesting
@@ -131,12 +137,18 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       throw new ClosedChannelException();
     }
 
-    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+    long begin = writeCtx.getConfirmedBytes().get();
+    currentContent = RewindableContent.of(srcs, srcsOffset, srcsLength);
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, finalize);
+    if (data.length == 0) {
+      currentContent = null;
+      return 0;
+    }
 
     List<BidiWriteObjectRequest> messages = new ArrayList<>();
 
-    int bytesConsumed = 0;
-    for (ChunkSegment datum : data) {
+    for (int i = 0; i < data.length; i++) {
+      ChunkSegment datum = data[i];
       Crc32cLengthKnown crc32c = datum.getCrc32c();
       ByteString b = datum.getB();
       int contentSize = b.size();
@@ -149,11 +161,14 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       if (crc32c != null) {
         checksummedData.setCrc32C(crc32c.getValue());
       }
-      BidiWriteObjectRequest.Builder builder =
-          writeCtx
-              .newRequestBuilder()
-              .setWriteOffset(offset)
-              .setChecksummedData(checksummedData.build());
+      BidiWriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+      if (!first) {
+        builder.clearUploadId();
+        builder.clearObjectChecksums();
+      } else {
+        first = false;
+      }
+      builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
       if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
         if (cumulative != null) {
@@ -163,10 +178,17 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
         finished = true;
       }
 
-      BidiWriteObjectRequest build = possiblyPairDownBidiRequest(builder, first).build();
-      first = false;
+      if (i == data.length - 1 && !finished) {
+        if (finalize) {
+          builder.setFinishWrite(true);
+          finished = true;
+        } else {
+          builder.setFlush(true).setStateLookup(true);
+        }
+      }
+
+      BidiWriteObjectRequest build = builder.build();
       messages.add(build);
-      bytesConsumed += contentSize;
     }
     if (finalize && !finished) {
       messages.add(finishMessage());
@@ -176,10 +198,14 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     try {
       flush(messages);
     } catch (RuntimeException e) {
+      open = false;
       resultFuture.setException(e);
       throw e;
     }
 
+    long end = writeCtx.getConfirmedBytes().get();
+
+    long bytesConsumed = end - begin;
     return bytesConsumed;
   }
 
@@ -188,8 +214,11 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     long offset = writeCtx.getTotalSentBytes().get();
     Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
 
-    BidiWriteObjectRequest.Builder b =
-        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    BidiWriteObjectRequest.Builder b = writeCtx.newRequestBuilder();
+    if (!first) {
+      b.clearUploadId().clearObjectChecksums();
+    }
+    b.setFinishWrite(true).setWriteOffset(offset);
     if (crc32cValue != null) {
       b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
     }
@@ -201,13 +230,9 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     if (stream == null) {
       synchronized (this) {
         if (stream == null) {
-          GrpcCallContext internalContext =
-              contextWithBucketName(bucketName, baseContextSupplier.get());
+          responseObserver.reset();
           stream =
-              this.write
-                  .withDefaultCallContext(internalContext)
-                  .bidiStreamingCall(responseObserver);
-          responseObserver.sem.drainPermits();
+              new GracefulOutboundStream(this.write.bidiStreamingCall(responseObserver, context));
         }
       }
     }
@@ -223,47 +248,28 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
             ApiStreamObserver<BidiWriteObjectRequest> opened = openedStream();
             for (BidiWriteObjectRequest message : segments) {
               opened.onNext(message);
+              lastWrittenRequest = message;
             }
-            if (!finished) {
-              BidiWriteObjectRequest message =
-                  BidiWriteObjectRequest.newBuilder().setFlush(true).setStateLookup(true).build();
-              opened.onNext(message);
+            if (lastWrittenRequest.getFinishWrite()) {
+              opened.onCompleted();
             }
             responseObserver.await();
             return null;
-          } catch (Exception e) {
+          } catch (Throwable t) {
             stream = null;
             first = true;
-            throw e;
+            t.addSuppressed(new AsyncStorageTaskException());
+            throw t;
           }
         },
         Decoder.identity());
   }
 
-  private static BidiWriteObjectRequest.Builder possiblyPairDownBidiRequest(
-      BidiWriteObjectRequest.Builder b, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && b.getWriteOffset() == 0) {
-      return b;
-    }
-
-    if (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
   private class BidiObserver implements ApiStreamObserver<BidiWriteObjectResponse> {
 
     private final Semaphore sem;
     private volatile BidiWriteObjectResponse last;
+    private volatile StorageException clientDetectedError;
     private volatile RuntimeException previousError;
 
     private BidiObserver() {
@@ -272,22 +278,96 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
 
     @Override
     public void onNext(BidiWriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        writeCtx.getConfirmedBytes().set((value.getPersistedSize()));
-      } else if (value.hasResource()) {
-        writeCtx.getConfirmedBytes().set(value.getResource().getSize());
+      boolean finalizing = lastWrittenRequest.getFinishWrite();
+      if (!finalizing && value.hasPersistedSize()) { // incremental
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long persistedSize = value.getPersistedSize();
+
+        if (totalSentBytes == persistedSize) {
+          writeCtx.getConfirmedBytes().set(persistedSize);
+          ok(value);
+        } else if (persistedSize < totalSentBytes) {
+          long delta = totalSentBytes - persistedSize;
+          // rewind our content and any state that my have run ahead of the actual ack'd bytes
+          currentContent.rewindTo(delta);
+          writeCtx.getTotalSentBytes().set(persistedSize);
+          writeCtx.getConfirmedBytes().set(persistedSize);
+          ok(value);
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_7.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else if (finalizing && value.hasResource()) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long finalSize = value.getResource().getSize();
+        if (totalSentBytes == finalSize) {
+          writeCtx.getConfirmedBytes().set(finalSize);
+          ok(value);
+        } else if (finalSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else if (!finalizing && value.hasResource()) {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_1.toStorageException(
+                ImmutableList.of(lastWrittenRequest), value, context, null));
+      } else if (finalizing && value.hasPersistedSize()) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long persistedSize = value.getPersistedSize();
+        // if a flush: true, state_lookup: true message is in the stream along with a
+        // finish_write: true, GCS can respond with the incremental update, gracefully handle this
+        // message
+        if (totalSentBytes == persistedSize) {
+          writeCtx.getConfirmedBytes().set(persistedSize);
+        } else if (persistedSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_3.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+                ImmutableList.of(lastWrittenRequest), value, context, null));
       }
-      sem.release();
-      last = value;
     }
 
     @Override
     public void onError(Throwable t) {
-      if (t instanceof RuntimeException) {
+      if (t instanceof OutOfRangeException) {
+        OutOfRangeException oore = (OutOfRangeException) t;
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                ImmutableList.of(lastWrittenRequest), null, context, oore));
+      } else if (t instanceof ApiException) {
+        // use StorageExceptions logic to translate from ApiException to our status codes ensuring
+        // things fall in line with our retry handlers.
+        // This is suboptimal, as it will initialize a second exception, however this is the
+        // unusual case, and it should not cause a significant overhead given its rarity.
+        StorageException tmp = StorageException.asStorageException((ApiException) t);
+        previousError =
+            ResumableSessionFailureScenario.toStorageException(
+                tmp.getCode(),
+                tmp.getMessage(),
+                tmp.getReason(),
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                t);
+        sem.release();
+      } else if (t instanceof RuntimeException) {
         previousError = (RuntimeException) t;
+        sem.release();
       }
-      sem.release();
     }
 
     @Override
@@ -298,6 +378,25 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       sem.release();
     }
 
+    private void ok(BidiWriteObjectResponse value) {
+      last = value;
+      sem.release();
+    }
+
+    private void clientDetectedError(StorageException storageException) {
+      open = false;
+      clientDetectedError = storageException;
+      // yes, check that previousError is not the same instance as e
+      if (previousError != null && previousError != storageException) {
+        storageException.addSuppressed(previousError);
+        previousError = null;
+      }
+      if (previousError == null) {
+        previousError = storageException;
+      }
+      sem.release();
+    }
+
     void await() {
       try {
         sem.acquire();
@@ -308,11 +407,69 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
           throw new RuntimeException(e);
         }
       }
+      StorageException e = clientDetectedError;
       RuntimeException err = previousError;
+      clientDetectedError = null;
+      previousError = null;
+      if ((e != null || err != null) && stream != null) {
+        if (lastWrittenRequest.getFinishWrite()) {
+          stream.onCompleted();
+        } else {
+          stream.onError(Status.CANCELLED.asRuntimeException());
+        }
+      }
+      if (e != null) {
+        throw e;
+      }
       if (err != null) {
-        previousError = null;
         throw err;
       }
     }
+
+    public void reset() {
+      sem.drainPermits();
+      last = null;
+      clientDetectedError = null;
+      previousError = null;
+    }
+  }
+
+  /**
+   * Prevent "already half-closed" if we previously called onComplete but then detect an error and
+   * call onError
+   */
+  private static final class GracefulOutboundStream
+      implements ApiStreamObserver<BidiWriteObjectRequest> {
+
+    private final ApiStreamObserver<BidiWriteObjectRequest> delegate;
+    private volatile boolean closing;
+
+    private GracefulOutboundStream(ApiStreamObserver<BidiWriteObjectRequest> delegate) {
+      this.delegate = delegate;
+      this.closing = false;
+    }
+
+    @Override
+    public void onNext(BidiWriteObjectRequest value) {
+      delegate.onNext(value);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      if (closing) {
+        return;
+      }
+      closing = true;
+      delegate.onError(t);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (closing) {
+        return;
+      }
+      closing = true;
+      delegate.onCompleted();
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
index 536eba7f..05387326 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
@@ -155,7 +155,7 @@ final class GapicBidiWritableByteChannelSessionBuilder {
                             resultFuture,
                             new ChunkSegmenter(
                                 boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-                            start,
+                            new BidiWriteCtx<>(start),
                             Retrying::newCallContext))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
index 16c6ec0a..984c7bfd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
@@ -20,11 +20,13 @@ import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
 
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -33,11 +35,7 @@ import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.ExecutionException;
-import java.util.function.Consumer;
-import java.util.function.LongConsumer;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
@@ -55,13 +53,14 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
   private boolean open = true;
   private boolean first = true;
   private boolean finished = false;
+  private volatile WriteObjectRequest lastWrittenRequest;
 
   GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      ResumableWrite requestFactory) {
-    String bucketName = requestFactory.bucketName();
+      WriteCtx<ResumableWrite> writeCtx) {
+    String bucketName = writeCtx.getRequestFactory().bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
@@ -69,8 +68,8 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
         contextWithBucketName(bucketName, GrpcCallContext.createDefault());
     this.write = write.withDefaultCallContext(internalContext);
 
-    this.writeCtx = new WriteCtx<>(requestFactory);
-    this.responseObserver = new Observer(writeCtx.getConfirmedBytes()::set, resultFuture::set);
+    this.writeCtx = writeCtx;
+    this.responseObserver = new Observer(internalContext);
   }
 
   @Override
@@ -92,27 +91,24 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
 
   @Override
   public void close() throws IOException {
+    if (!open) {
+      return;
+    }
+    open = false;
     ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
-    if (!finished) {
-      WriteObjectRequest message = finishMessage();
-      try {
+    try {
+      if (!finished) {
+        WriteObjectRequest message = finishMessage();
+        lastWrittenRequest = message;
         openedStream.onNext(message);
-        openedStream.onCompleted();
         finished = true;
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
-      }
-    } else {
-      try {
-        openedStream.onCompleted();
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
       }
+      openedStream.onCompleted();
+      responseObserver.await();
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
     }
-    open = false;
-    responseObserver.await();
   }
 
   private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
@@ -122,51 +118,54 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
     }
 
     ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
-
-    List<WriteObjectRequest> messages = new ArrayList<>();
+    if (data.length == 0) {
+      return 0;
+    }
 
     ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
     int bytesConsumed = 0;
-    for (ChunkSegment datum : data) {
-      Crc32cLengthKnown crc32c = datum.getCrc32c();
-      ByteString b = datum.getB();
-      int contentSize = b.size();
-      long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
-      Crc32cLengthKnown cumulative =
-          writeCtx
-              .getCumulativeCrc32c()
-              .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
-      ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
-      if (crc32c != null) {
-        checksummedData.setCrc32C(crc32c.getValue());
-      }
-      WriteObjectRequest.Builder builder =
-          writeCtx
-              .newRequestBuilder()
-              .setWriteOffset(offset)
-              .setChecksummedData(checksummedData.build());
-      if (!datum.isOnlyFullBlocks()) {
-        builder.setFinishWrite(true);
-        if (cumulative != null) {
-          builder.setObjectChecksums(
-              ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+    try {
+      for (int i = 0; i < data.length; i++) {
+        ChunkSegment datum = data[i];
+        Crc32cLengthKnown crc32c = datum.getCrc32c();
+        ByteString b = datum.getB();
+        int contentSize = b.size();
+        long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+        Crc32cLengthKnown cumulative =
+            writeCtx
+                .getCumulativeCrc32c()
+                .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
+        ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
+        if (crc32c != null) {
+          checksummedData.setCrc32C(crc32c.getValue());
+        }
+        WriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+        if (!first) {
+          builder.clearUploadId();
+          builder.clearWriteObjectSpec();
+          builder.clearObjectChecksums();
+        }
+        builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
+        if (!datum.isOnlyFullBlocks() || (finalize && i + 1 == data.length)) {
+          builder.setFinishWrite(true);
+          if (cumulative != null) {
+            builder.setObjectChecksums(
+                ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+          }
+          finished = true;
         }
-        finished = true;
-      }
-
-      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
-      first = false;
-      messages.add(build);
-      bytesConsumed += contentSize;
-    }
-    if (finalize && !finished) {
-      messages.add(finishMessage());
-      finished = true;
-    }
 
-    try {
-      for (WriteObjectRequest message : messages) {
-        openedStream.onNext(message);
+        WriteObjectRequest build = builder.build();
+        first = false;
+        lastWrittenRequest = build;
+        openedStream.onNext(build);
+        bytesConsumed += contentSize;
+      }
+      if (finalize && !finished) {
+        WriteObjectRequest finishMessage = finishMessage();
+        lastWrittenRequest = finishMessage;
+        openedStream.onNext(finishMessage);
+        finished = true;
       }
     } catch (RuntimeException e) {
       resultFuture.setException(e);
@@ -201,73 +200,104 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
     return stream;
   }
 
-  /**
-   * Several fields of a WriteObjectRequest are only allowed on the "first" message sent to gcs,
-   * this utility method centralizes the logic necessary to clear those fields for use by subsequent
-   * messages.
-   */
-  private static WriteObjectRequest.Builder possiblyPairDownRequest(
-      WriteObjectRequest.Builder b, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && b.getWriteOffset() == 0) {
-      return b;
-    }
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
-  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+  class Observer implements ApiStreamObserver<WriteObjectResponse> {
 
-    private final LongConsumer sizeCallback;
-    private final Consumer<WriteObjectResponse> completeCallback;
+    private final GrpcCallContext context;
 
     private final SettableApiFuture<Void> invocationHandle;
     private volatile WriteObjectResponse last;
 
-    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
+    Observer(GrpcCallContext context) {
+      this.context = context;
       this.invocationHandle = SettableApiFuture.create();
     }
 
     @Override
     public void onNext(WriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        sizeCallback.accept(value.getPersistedSize());
-      } else if (value.hasResource()) {
-        sizeCallback.accept(value.getResource().getSize());
-      }
       last = value;
     }
 
-    /**
-     * observed exceptions so far
-     *
-     * <ol>
-     *   <li>{@link com.google.api.gax.rpc.OutOfRangeException}
-     *   <li>{@link com.google.api.gax.rpc.AlreadyExistsException}
-     *   <li>{@link io.grpc.StatusRuntimeException}
-     * </ol>
-     */
     @Override
     public void onError(Throwable t) {
-      invocationHandle.setException(t);
+      if (t instanceof ApiException) {
+        // use StorageExceptions logic to translate from ApiException to our status codes ensuring
+        // things fall in line with our retry handlers.
+        // This is suboptimal, as it will initialize a second exception, however this is the
+        // unusual case, and it should not cause a significant overhead given its rarity.
+        StorageException tmp = StorageException.asStorageException((ApiException) t);
+        StorageException storageException =
+            ResumableSessionFailureScenario.toStorageException(
+                tmp.getCode(),
+                tmp.getMessage(),
+                tmp.getReason(),
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                t);
+        resultFuture.setException(storageException);
+        invocationHandle.setException(storageException);
+      } else {
+        resultFuture.setException(t);
+        invocationHandle.setException(t);
+      }
     }
 
     @Override
     public void onCompleted() {
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
+      boolean finalizing = lastWrittenRequest.getFinishWrite();
+      if (last == null) {
+        clientDetectedError(
+            ResumableSessionFailureScenario.toStorageException(
+                0,
+                "onComplete without preceding onNext, unable to determine success.",
+                "invalid",
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                null));
+      } else if (last.hasResource() /* && finalizing*/) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long finalSize = last.getResource().getSize();
+        if (totalSentBytes == finalSize) {
+          ok(finalSize);
+        } else if (finalSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), last, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), last, context, null));
+        }
+      } else if (!finalizing || last.hasPersistedSize()) { // unexpected incremental response
+        clientDetectedError(
+            ResumableSessionFailureScenario.toStorageException(
+                0,
+                "Unexpected incremental response for finalizing request.",
+                "invalid",
+                ImmutableList.of(lastWrittenRequest),
+                last,
+                context,
+                null));
+      } else {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+                ImmutableList.of(lastWrittenRequest), last, context, null));
       }
+    }
+
+    private void ok(long persistedSize) {
+      writeCtx.getConfirmedBytes().set(persistedSize);
+      resultFuture.set(last);
       invocationHandle.set(null);
     }
 
+    private void clientDetectedError(StorageException storageException) {
+      open = false;
+      resultFuture.setException(storageException);
+      invocationHandle.setException(storageException);
+    }
+
     void await() {
       try {
         invocationHandle.get();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index f4a8afdb..1e02f968 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -305,7 +305,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, start);
+                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
                 })
                 .andThen(StorageByteChannels.writable()::createSynchronized));
@@ -346,7 +346,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, start);
+                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
                 })
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index c195ce78..0d0904ca 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -796,7 +796,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     //  3. wrap the result in another future container before constructing the BlobWriteChannel
     ApiFuture<ResumableWrite> wrapped = ApiFutures.immediateFuture(resumableWrite);
     return new GrpcBlobWriteChannel(
-        storageClient.writeObjectCallable(),
+        storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext),
         getOptions(),
         retryAlgorithmManager.idempotent(),
         () -> wrapped,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 38acf828..92081e03 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -600,7 +600,13 @@ public final class GrpcStorageOptions extends StorageOptions
     @BetaApi
     @Override
     public GrpcStorageOptions build() {
-      return new GrpcStorageOptions(this, defaults());
+      GrpcStorageOptions options = new GrpcStorageOptions(this, defaults());
+      // TODO: Remove when https://github.com/googleapis/sdk-platform-java/issues/2911 is resolved
+      if (options.getUniverseDomain() != null) {
+        this.setHost("https://storage." + options.getUniverseDomain());
+        return new GrpcStorageOptions(this, defaults());
+      }
+      return options;
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
index 31784bfd..7d0e46ce 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
@@ -23,6 +23,7 @@ import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiExceptions;
+import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.RecoveryFileManager.RecoveryVolumeSinkFactory;
 import com.google.cloud.storage.Storage.BlobWriteOption;
@@ -34,6 +35,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.ServiceConstants.Values;
+import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -184,16 +186,20 @@ public final class JournalingBlobWriteSessionConfig extends BlobWriteSessionConf
       if (storage instanceof GrpcStorageImpl) {
         GrpcStorageImpl grpcStorage = (GrpcStorageImpl) storage;
         RecoveryFile recoveryFile = recoveryFileManager.newRecoveryFile(info);
+        GrpcCallContext grpcCallContext =
+            opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
         ApiFuture<ResumableWrite> f =
             grpcStorage.startResumableWrite(
-                GrpcCallContext.createDefault(), grpcStorage.getWriteObjectRequest(info, opts));
+                grpcCallContext, grpcStorage.getWriteObjectRequest(info, opts));
         ApiFuture<WriteCtx<ResumableWrite>> start =
             ApiFutures.transform(f, WriteCtx::new, MoreExecutors.directExecutor());
 
+        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
+            grpcStorage.storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext);
         BufferedWritableByteChannelSession<WriteObjectResponse> session =
             ResumableMedia.gapic()
                 .write()
-                .byteChannel(grpcStorage.storageClient.writeObjectCallable())
+                .byteChannel(write)
                 .setHasher(Hasher.noop())
                 .setByteStringStrategy(ByteStringStrategy.copy())
                 .journaling()
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
index 8c06d70d..88dbaf7b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
@@ -26,11 +26,11 @@ import com.google.api.gax.rpc.ApiException;
 import com.google.cloud.BaseServiceException;
 import com.google.cloud.storage.StorageException.IOExceptionCallable;
 import com.google.common.io.CharStreams;
-import com.google.protobuf.MessageOrBuilder;
+import com.google.protobuf.Message;
+import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
 import com.google.storage.v2.WriteObjectRequest;
-import com.google.storage.v2.WriteObjectResponse;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -81,10 +81,6 @@ enum ResumableSessionFailureScenario {
   private static final String PREFIX_I = "\t|< ";
   private static final String PREFIX_O = "\t|> ";
   private static final String PREFIX_X = "\t|  ";
-  // define some constants for tab widths that are more compressed that the literals
-  private static final String T1 = "\t";
-  private static final String T2 = "\t\t";
-  private static final String T3 = "\t\t\t";
 
   private static final Predicate<String> includedHeaders =
       matches("Content-Length")
@@ -134,8 +130,10 @@ enum ResumableSessionFailureScenario {
   }
 
   StorageException toStorageException(
-      @NonNull List<@NonNull WriteObjectRequest> reqs,
-      @Nullable WriteObjectResponse resp,
+      /*In java List<WriteObjectRequest> is not a sub-type of List<Message> despite WriteObjectRequest being a Message.
+       * intentionally only define List so the compiler doesn't complain */
+      @SuppressWarnings("rawtypes") @NonNull List reqs,
+      @Nullable Message resp,
       @NonNull GrpcCallContext context,
       @Nullable Throwable cause) {
     return toStorageException(code, message, reason, reqs, resp, context, cause);
@@ -161,8 +159,8 @@ enum ResumableSessionFailureScenario {
       int code,
       String message,
       @Nullable String reason,
-      @NonNull List<@NonNull WriteObjectRequest> reqs,
-      @Nullable WriteObjectResponse resp,
+      @NonNull List reqs,
+      @Nullable Message resp,
       @NonNull GrpcCallContext context,
       @Nullable Throwable cause) {
     final StringBuilder sb = new StringBuilder();
@@ -177,35 +175,8 @@ enum ResumableSessionFailureScenario {
       } else {
         sb.append(",");
       }
-      WriteObjectRequest req = reqs.get(i);
-      sb.append("\n").append(PREFIX_O).append(T1).append(req.getClass().getName()).append("{");
-      if (req.hasUploadId()) {
-        sb.append("\n").append(PREFIX_O).append(T2).append("upload_id: ").append(req.getUploadId());
-      }
-      long writeOffset = req.getWriteOffset();
-      if (req.hasChecksummedData()) {
-        ChecksummedData checksummedData = req.getChecksummedData();
-        sb.append("\n").append(PREFIX_O).append(T2);
-        sb.append(
-            String.format(
-                "checksummed_data: {range: [%d:%d]",
-                writeOffset, writeOffset + checksummedData.getContent().size()));
-        if (checksummedData.hasCrc32C()) {
-          sb.append(", crc32c: ").append(checksummedData.getCrc32C());
-        }
-        sb.append("}");
-      } else {
-        sb.append("\n").append(PREFIX_O).append(T2).append("write_offset: ").append(writeOffset);
-      }
-      if (req.getFinishWrite()) {
-        sb.append("\n").append(PREFIX_O).append(T2).append("finish_write: true");
-      }
-      if (req.hasObjectChecksums()) {
-        ObjectChecksums objectChecksums = req.getObjectChecksums();
-        sb.append("\n").append(PREFIX_O).append(T2).append("object_checksums: ").append("{");
-        fmt(objectChecksums, PREFIX_O, T3, sb);
-        sb.append("\n").append(PREFIX_O).append(T2).append("}");
-      }
+      Message req = (Message) reqs.get(i);
+      fmt(req, PREFIX_O, Indentation.T1, sb);
       sb.append("\n").append(PREFIX_O).append("\t}");
       if (i == length - 1) {
         sb.append("\n").append(PREFIX_O).append("]");
@@ -217,7 +188,7 @@ enum ResumableSessionFailureScenario {
     // response context
     if (resp != null) {
       sb.append("\n").append(PREFIX_I).append(resp.getClass().getName()).append("{");
-      fmt(resp, PREFIX_I, T1, sb);
+      fmt(resp, PREFIX_I, Indentation.T1, sb);
       sb.append("\n").append(PREFIX_I).append("}");
       sb.append("\n").append(PREFIX_X);
     }
@@ -250,7 +221,8 @@ enum ResumableSessionFailureScenario {
         sb.append("\n").append(PREFIX_X);
       }
     }
-    return new StorageException(code, sb.toString(), reason, cause);
+    StorageException se = new StorageException(code, sb.toString(), reason, cause);
+    return se;
   }
 
   static StorageException toStorageException(
@@ -359,16 +331,122 @@ enum ResumableSessionFailureScenario {
   }
 
   private static void fmt(
-      MessageOrBuilder msg,
+      Message msg,
       @SuppressWarnings("SameParameterValue") String prefix,
-      String indentation,
+      Indentation indentation,
       StringBuilder sb) {
-    String string = msg.toString();
-    // drop the final new line before prefixing
-    string = string.replaceAll("\n$", "");
-    sb.append("\n")
-        .append(prefix)
-        .append(indentation)
-        .append(string.replaceAll("\r?\n", "\n" + prefix + indentation));
+    if (msg instanceof WriteObjectRequest) {
+      WriteObjectRequest req = (WriteObjectRequest) msg;
+      fmtWriteObjectRequest(req, prefix, indentation, sb);
+    } else if (msg instanceof BidiWriteObjectRequest) {
+      BidiWriteObjectRequest req = (BidiWriteObjectRequest) msg;
+      fmtBidiWriteObjectRequest(req, prefix, indentation, sb);
+    } else {
+      String string = msg.toString();
+      // drop the final new line before prefixing
+      string = string.replaceAll("\n$", "");
+      sb.append("\n")
+          .append(prefix)
+          .append(indentation)
+          .append(string.replaceAll("\r?\n", "\n" + prefix + indentation.indentation));
+    }
+  }
+
+  private static void fmtWriteObjectRequest(
+      WriteObjectRequest req, String prefix, Indentation t1, StringBuilder sb) {
+    Indentation t2 = t1.indent();
+    Indentation t3 = t2.indent();
+    sb.append("\n").append(prefix).append(t1).append(req.getClass().getName()).append("{");
+    if (req.hasUploadId()) {
+      sb.append("\n").append(prefix).append(t2).append("upload_id: ").append(req.getUploadId());
+    }
+    long writeOffset = req.getWriteOffset();
+    if (req.hasChecksummedData()) {
+      ChecksummedData checksummedData = req.getChecksummedData();
+      sb.append("\n").append(prefix).append(t2);
+      sb.append(
+          String.format(
+              "checksummed_data: {range: [%d:%d]",
+              writeOffset, writeOffset + checksummedData.getContent().size()));
+      if (checksummedData.hasCrc32C()) {
+        sb.append(", crc32c: ").append(checksummedData.getCrc32C());
+      }
+      sb.append("}");
+    } else {
+      sb.append("\n").append(prefix).append(t2).append("write_offset: ").append(writeOffset);
+    }
+    if (req.getFinishWrite()) {
+      sb.append("\n").append(prefix).append(t2).append("finish_write: true");
+    }
+    if (req.hasObjectChecksums()) {
+      ObjectChecksums objectChecksums = req.getObjectChecksums();
+      sb.append("\n").append(prefix).append(t2).append("object_checksums: ").append("{");
+      fmt(objectChecksums, prefix, t3, sb);
+      sb.append("\n").append(prefix).append(t2).append("}");
+    }
+  }
+
+  private static void fmtBidiWriteObjectRequest(
+      BidiWriteObjectRequest req, String prefix, Indentation t1, StringBuilder sb) {
+    Indentation t2 = t1.indent();
+    Indentation t3 = t2.indent();
+    sb.append("\n").append(prefix).append(t1).append(req.getClass().getName()).append("{");
+    if (req.hasUploadId()) {
+      sb.append("\n").append(prefix).append(t2).append("upload_id: ").append(req.getUploadId());
+    }
+    long writeOffset = req.getWriteOffset();
+    if (req.hasChecksummedData()) {
+      ChecksummedData checksummedData = req.getChecksummedData();
+      sb.append("\n").append(prefix).append(t2);
+      sb.append(
+          String.format(
+              "checksummed_data: {range: [%d:%d]",
+              writeOffset, writeOffset + checksummedData.getContent().size()));
+      if (checksummedData.hasCrc32C()) {
+        sb.append(", crc32c: ").append(checksummedData.getCrc32C());
+      }
+      sb.append("}");
+    } else {
+      sb.append("\n").append(prefix).append(t2).append("write_offset: ").append(writeOffset);
+    }
+    if (req.getFlush()) {
+      sb.append("\n").append(prefix).append(t2).append("flush: true");
+    }
+    if (req.getStateLookup()) {
+      sb.append("\n").append(prefix).append(t2).append("state_lookup: true");
+    }
+    if (req.getFinishWrite()) {
+      sb.append("\n").append(prefix).append(t2).append("finish_write: true");
+    }
+    if (req.hasObjectChecksums()) {
+      ObjectChecksums objectChecksums = req.getObjectChecksums();
+      sb.append("\n").append(prefix).append(t2).append("object_checksums: ").append("{");
+      fmt(objectChecksums, prefix, t3, sb);
+      sb.append("\n").append(prefix).append(t2).append("}");
+    }
+  }
+
+  enum Indentation {
+    T1("\t"),
+    T2("\t\t"),
+    T3("\t\t\t"),
+    T4("\t\t\t\t"),
+    ;
+
+    private final String indentation;
+
+    Indentation(String indentation) {
+      this.indentation = indentation;
+    }
+
+    Indentation indent() {
+      int ordinal = ordinal();
+      return values()[ordinal + 1];
+    }
+
+    @Override
+    public String toString() {
+      return indentation;
+    }
   }
 }
diff --git a/grpc-google-cloud-storage-control-v2/pom.xml b/grpc-google-cloud-storage-control-v2/pom.xml
index 248d2d8b..1b571e92 100644
--- a/grpc-google-cloud-storage-control-v2/pom.xml
+++ b/grpc-google-cloud-storage-control-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-  <version>2.40.0</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
   <name>grpc-google-cloud-storage-control-v2</name>
   <description>GRPC library for google-cloud-storage</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/grpc-google-cloud-storage-v2/pom.xml b/grpc-google-cloud-storage-v2/pom.xml
index 44bd9d5a..0e7be55a 100644
--- a/grpc-google-cloud-storage-v2/pom.xml
+++ b/grpc-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-v2</artifactId>
-  <version>2.40.0-alpha</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+  <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
   <name>grpc-google-cloud-storage-v2</name>
   <description>GRPC library for grpc-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/pom.xml b/pom.xml
index 306f89a6..280a4196 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-parent</artifactId>
   <packaging>pom</packaging>
-  <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -76,12 +76,12 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.apis</groupId>
         <artifactId>google-api-services-storage</artifactId>
-        <version>v1-rev20240319-2.0.0</version>
+        <version>v1-rev20240524-2.0.0</version>
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
@@ -98,32 +98,32 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>gapic-google-cloud-storage-v2</artifactId>
-        <version>2.40.0-alpha</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
+        <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
diff --git a/proto-google-cloud-storage-control-v2/pom.xml b/proto-google-cloud-storage-control-v2/pom.xml
index c677dff1..24bdbda7 100644
--- a/proto-google-cloud-storage-control-v2/pom.xml
+++ b/proto-google-cloud-storage-control-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-  <version>2.40.0</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+  <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
   <name>proto-google-cloud-storage-control-v2</name>
   <description>Proto library for  proto-google-cloud-storage-control-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/proto-google-cloud-storage-v2/pom.xml b/proto-google-cloud-storage-v2/pom.xml
index 3a7df5c8..b8472faf 100644
--- a/proto-google-cloud-storage-v2/pom.xml
+++ b/proto-google-cloud-storage-v2/pom.xml
@@ -4,13 +4,13 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-v2</artifactId>
-  <version>2.40.0-alpha</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
+  <version>2.40.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-v2:current} -->
   <name>proto-google-cloud-storage-v2</name>
   <description>PROTO library for proto-google-cloud-storage-v2</description>
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <dependencies>
     <dependency>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 6fd6fa51..e377f15c 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,12 +30,12 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.39.0</version>
+      <version>2.40.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index ef08c54f..2e9a84a9 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -67,7 +67,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 65f639d6..e1faf99a 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,12 +28,12 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
 
diff --git a/storage-shared-benchmarking/pom.xml b/storage-shared-benchmarking/pom.xml
index 98c9ff31..98e8e19f 100644
--- a/storage-shared-benchmarking/pom.xml
+++ b/storage-shared-benchmarking/pom.xml
@@ -10,7 +10,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
 
   <properties>
@@ -31,7 +31,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.40.0</version><!-- {x-version-update:google-cloud-storage:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
       <classifier>tests</classifier>
     </dependency>
     <dependency>
diff --git a/versions.txt b/versions.txt
index b491c733..39d03adf 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,10 +1,10 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:2.40.0:2.40.0
-gapic-google-cloud-storage-v2:2.40.0-alpha:2.40.0-alpha
-grpc-google-cloud-storage-v2:2.40.0-alpha:2.40.0-alpha
-proto-google-cloud-storage-v2:2.40.0-alpha:2.40.0-alpha
-google-cloud-storage-control:2.40.0:2.40.0
-proto-google-cloud-storage-control-v2:2.40.0:2.40.0
-grpc-google-cloud-storage-control-v2:2.40.0:2.40.0
+google-cloud-storage:2.40.0:2.40.1-SNAPSHOT
+gapic-google-cloud-storage-v2:2.40.0-alpha:2.40.1-alpha-SNAPSHOT
+grpc-google-cloud-storage-v2:2.40.0-alpha:2.40.1-alpha-SNAPSHOT
+proto-google-cloud-storage-v2:2.40.0-alpha:2.40.1-alpha-SNAPSHOT
+google-cloud-storage-control:2.40.0:2.40.1-SNAPSHOT
+proto-google-cloud-storage-control-v2:2.40.0:2.40.1-SNAPSHOT
+grpc-google-cloud-storage-control-v2:2.40.0:2.40.1-SNAPSHOT


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java
index d2778342..578f6c7d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeServer.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.it.GrpcPlainRequestLoggingInterceptor;
 import com.google.storage.v2.StorageGrpc;
 import com.google.storage.v2.StorageSettings;
 import io.grpc.Server;
@@ -58,6 +59,7 @@ final class FakeServer implements AutoCloseable {
             .setHost("http://" + endpoint)
             .setProjectId("test-proj")
             .setCredentials(NoCredentials.getInstance())
+            .setGrpcInterceptorProvider(GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
             .build();
     return new FakeServer(server, grpcStorageOptions);
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicBidiUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicBidiUnbufferedWritableByteChannelTest.java
new file mode 100644
index 00000000..7c3fb1fd
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicBidiUnbufferedWritableByteChannelTest.java
@@ -0,0 +1,938 @@
+/*
+ * Copyright 2024 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._256KiB;
+import static com.google.cloud.storage.ByteSizeConstants._512KiB;
+import static com.google.cloud.storage.ByteSizeConstants._768KiB;
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.BidiWriteObjectResponse;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageGrpc.StorageImplBase;
+import io.grpc.Status.Code;
+import io.grpc.stub.CallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.logging.Logger;
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.junit.Test;
+
+public final class ITGapicBidiUnbufferedWritableByteChannelTest {
+
+  private static final ChunkSegmenter CHUNK_SEGMENTER =
+      new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), _256KiB, _256KiB);
+
+  /**
+   *
+   *
+   * <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, 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>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset= 0, checksummed_data.content.length = 262144 }
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ resources = {name = obj, size = 525288 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario1() throws Exception {
+
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .setStateLookup(true)
+            .setFlush(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("obj").setSize(_512KiB).build())
+            .build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      ByteBuffer bb = DataGenerator.base64Characters().genByteBuffer(_256KiB);
+      StorageException se = assertThrows(StorageException.class, () -> channel.write(bb));
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("invalid"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 262144, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ persisted_size = 525288 })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario2() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setFinishWrite(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder().setPersistedSize(_512KiB).build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(_256KiB);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("invalid"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_256KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ persisted_size = 262144 })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario3() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder().setPersistedSize(_256KiB).build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(_512KiB);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_512KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 262144}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 262144, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 262144, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ resources = {name = obj, size = 262144 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setFinishWrite(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_256KiB).build())
+            .build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(_256KiB);
+
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      channel.close();
+
+      BidiWriteObjectResponse BidiWriteObjectResponse = done.get(2, TimeUnit.SECONDS);
+      assertThat(BidiWriteObjectResponse).isEqualTo(resp1);
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 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>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ resources = {name = obj, size = 262144 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_1() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_256KiB).build())
+            .build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(_512KiB);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_512KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 786432}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(BidiWriteObjectResponse{ resources = {name = obj, size = 786432 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_2() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_768KiB).build())
+            .build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(_512KiB);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_512KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 0
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 262144, data = [262144:524288]
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * BidiWriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 262144, checksummed_data.content.length = 262144}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onError(Status{code=OUT_OF_RANGE, description="Upload request started at offset '262144', which is past expected offset '0'."})
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario5() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB))))
+            .setStateLookup(true)
+            .setFlush(true)
+            .build();
+    StorageImplBase service1 =
+        new BidiWriteService(
+            (obs, requests) -> {
+              if (requests.equals(ImmutableList.of(req1))) {
+                obs.onError(
+                    TestUtils.apiException(
+                        Code.OUT_OF_RANGE,
+                        "Upload request started at offset '262144', which is past expected offset '0'."));
+              } else {
+                obs.onError(
+                    TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
+              }
+            });
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(_256KiB);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      ByteBuffer bb = DataGenerator.base64Characters().genByteBuffer(_256KiB);
+      StorageException se = assertThrows(StorageException.class, () -> channel.write(bb));
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_256KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 {
+
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .setStateLookup(true)
+            .setFlush(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder().setPersistedSize(_512KiB).build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      ByteBuffer buf = DataGenerator.base64Characters().genByteBuffer(_256KiB);
+      StorageException se = assertThrows(StorageException.class, () -> channel.write(buf));
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  @Test
+  public void incremental_success() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .setStateLookup(true)
+            .setFlush(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder().setPersistedSize(_256KiB).build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              CHUNK_SEGMENTER,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      ByteBuffer buf = DataGenerator.base64Characters().genByteBuffer(_256KiB);
+      int written = channel.write(buf);
+      assertAll(
+          () -> assertThat(buf.remaining()).isEqualTo(0),
+          () -> assertThat(written).isEqualTo(_256KiB),
+          () -> assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(_256KiB),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_256KiB));
+    }
+  }
+
+  @Test
+  public void incremental_partialSuccess() throws Exception {
+    String uploadId = "uploadId";
+    BidiWriteObjectRequest req1 =
+        BidiWriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_512KiB)))
+                    .build())
+            .setStateLookup(true)
+            .setFlush(true)
+            .build();
+    BidiWriteObjectResponse resp1 =
+        BidiWriteObjectResponse.newBuilder().setPersistedSize(_256KiB).build();
+
+    ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    BidiWriteService service1 = new BidiWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<BidiWriteObjectResponse> done = SettableApiFuture.create();
+      BidiResumableWrite resumableWrite = getResumableWrite(uploadId);
+      BidiWriteCtx<BidiResumableWrite> writeCtx = new BidiWriteCtx<>(resumableWrite);
+
+      ChunkSegmenter chunkSegmenter =
+          new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), _512KiB, _256KiB);
+      //noinspection resource
+      GapicBidiUnbufferedWritableByteChannel channel =
+          new GapicBidiUnbufferedWritableByteChannel(
+              storageClient.bidiWriteObjectCallable(),
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              done,
+              chunkSegmenter,
+              writeCtx,
+              GrpcCallContext::createDefault);
+
+      ByteBuffer buf = DataGenerator.base64Characters().genByteBuffer(_512KiB);
+      int written = channel.write(buf);
+      assertAll(
+          () -> assertThat(buf.remaining()).isEqualTo(_256KiB),
+          () -> assertThat(written).isEqualTo(_256KiB),
+          () ->
+              assertWithMessage("totalSentBytes")
+                  .that(writeCtx.getTotalSentBytes().get())
+                  .isEqualTo(_256KiB),
+          () ->
+              assertWithMessage("confirmedBytes")
+                  .that(writeCtx.getConfirmedBytes().get())
+                  .isEqualTo(_256KiB));
+    }
+  }
+
+  private static @NonNull BidiResumableWrite getResumableWrite(String uploadId) {
+    StartResumableWriteRequest req = StartResumableWriteRequest.getDefaultInstance();
+    StartResumableWriteResponse resp =
+        StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
+    return new BidiResumableWrite(
+        req, resp, id -> BidiWriteObjectRequest.newBuilder().setUploadId(id).build());
+  }
+
+  static class BidiWriteService extends StorageImplBase {
+    private static final Logger LOGGER = Logger.getLogger(BidiWriteService.class.getName());
+    private final BiConsumer<StreamObserver<BidiWriteObjectResponse>, List<BidiWriteObjectRequest>>
+        c;
+
+    private ImmutableList.Builder<BidiWriteObjectRequest> requests;
+
+    BidiWriteService(
+        BiConsumer<StreamObserver<BidiWriteObjectResponse>, List<BidiWriteObjectRequest>> c) {
+      this.c = c;
+      this.requests = new ImmutableList.Builder<>();
+    }
+
+    BidiWriteService(ImmutableMap<List<BidiWriteObjectRequest>, BidiWriteObjectResponse> writes) {
+      this(
+          (obs, build) -> {
+            if (writes.containsKey(build)) {
+              obs.onNext(writes.get(build));
+              last(build)
+                  .filter(BidiWriteObjectRequest::getFinishWrite)
+                  .ifPresent(ignore -> obs.onCompleted());
+            } else {
+              logUnexpectedRequest(writes.keySet(), build);
+              obs.onError(
+                  TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
+            }
+          });
+    }
+
+    private static <T> Optional<T> last(List<T> l) {
+      if (l.isEmpty()) {
+        return Optional.empty();
+      } else {
+        return Optional.of(l.get(l.size() - 1));
+      }
+    }
+
+    private static void logUnexpectedRequest(
+        Set<List<BidiWriteObjectRequest>> writes, List<BidiWriteObjectRequest> build) {
+      Collector<CharSequence, ?, String> joining = Collectors.joining(",\n\t", "[\n\t", "\n]");
+      Collector<CharSequence, ?, String> oneLine = Collectors.joining(",", "[", "]");
+      String msg =
+          String.format(
+              "Unexpected Request Chain.%nexpected one of: %s%n        but was: %s",
+              writes.stream()
+                  .map(l -> l.stream().map(StorageV2ProtoUtils::fmtProto).collect(oneLine))
+                  .collect(joining),
+              build.stream().map(StorageV2ProtoUtils::fmtProto).collect(oneLine));
+      LOGGER.warning(msg);
+    }
+
+    @Override
+    public StreamObserver<BidiWriteObjectRequest> bidiWriteObject(
+        StreamObserver<BidiWriteObjectResponse> obs) {
+      return new Adapter() {
+        @Override
+        public void onNext(BidiWriteObjectRequest value) {
+          requests.add(value);
+          if ((value.getFlush() && value.getStateLookup()) || value.getFinishWrite()) {
+            ImmutableList<BidiWriteObjectRequest> build = requests.build();
+            c.accept(obs, build);
+          }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+          requests = new ImmutableList.Builder<>();
+        }
+
+        @Override
+        public void onCompleted() {
+          requests = new ImmutableList.Builder<>();
+        }
+      };
+    }
+  }
+
+  private abstract static class Adapter extends CallStreamObserver<BidiWriteObjectRequest> {
+
+    private Adapter() {}
+
+    @Override
+    public boolean isReady() {
+      return true;
+    }
+
+    @Override
+    public void setOnReadyHandler(Runnable onReadyHandler) {}
+
+    @Override
+    public void disableAutoInboundFlowControl() {}
+
+    @Override
+    public void request(int count) {}
+
+    @Override
+    public void setMessageCompression(boolean enable) {}
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedFinalizeOnCloseResumableWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedFinalizeOnCloseResumableWritableByteChannelTest.java
new file mode 100644
index 00000000..8181bd2b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedFinalizeOnCloseResumableWritableByteChannelTest.java
@@ -0,0 +1,329 @@
+/*
+ * Copyright 2024 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._256KiB;
+import static com.google.cloud.storage.ByteSizeConstants._512KiB;
+import static com.google.cloud.storage.ByteSizeConstants._768KiB;
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.cloud.storage.ITGapicUnbufferedWritableByteChannelTest.DirectWriteService;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.junit.Test;
+
+public final class ITGapicUnbufferedFinalizeOnCloseResumableWritableByteChannelTest {
+
+  private static final ChunkSegmenter CHUNK_SEGMENTER =
+      new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), _256KiB, _256KiB);
+
+  @Test
+  public void incrementalResponseForFinalizingRequest() throws Exception {
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    WriteObjectResponse resp1 = WriteObjectResponse.newBuilder().setPersistedSize(_256KiB).build();
+
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    DirectWriteService service1 = new DirectWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<WriteObjectResponse> done = SettableApiFuture.create();
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      //noinspection resource
+      GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel channel =
+          new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      se.printStackTrace(System.out);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("invalid"),
+          () -> assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(_512KiB),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 262144}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 262144, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * writeObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 262144, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ resources = {name = obj, size = 262144 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4() throws Exception {
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setFinishWrite(true)
+            .build();
+    WriteObjectResponse resp1 =
+        WriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_256KiB).build())
+            .build();
+
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    DirectWriteService service1 = new DirectWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<WriteObjectResponse> done = SettableApiFuture.create();
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel channel =
+          new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      channel.close();
+
+      WriteObjectResponse writeObjectResponse = done.get(2, TimeUnit.SECONDS);
+      assertThat(writeObjectResponse).isEqualTo(resp1);
+      assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_256KiB);
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 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>
+   * writeObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ resources = {name = obj, size = 262144 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_1() throws Exception {
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    WriteObjectResponse resp1 =
+        WriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_256KiB).build())
+            .build();
+
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    DirectWriteService service1 = new DirectWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<WriteObjectResponse> done = SettableApiFuture.create();
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      //noinspection resource
+      GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel channel =
+          new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      se.printStackTrace(System.out);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /**
+   *
+   *
+   * <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 = 786432}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>client state</td>
+   *     <td><pre>
+   * write_offset = 524288, finish = true
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>request</td>
+   *     <td><pre>
+   * writeObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ resources = {name = obj, size = 786432 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario4_2() throws Exception {
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_512KiB)
+            .setFinishWrite(true)
+            .build();
+    WriteObjectResponse resp1 =
+        WriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("name").setSize(_768KiB).build())
+            .build();
+
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> map =
+        ImmutableMap.of(ImmutableList.of(req1), resp1);
+    DirectWriteService service1 = new DirectWriteService(map);
+
+    try (FakeServer fakeServer = FakeServer.of(service1);
+        GrpcStorageImpl storage =
+            (GrpcStorageImpl) fakeServer.getGrpcStorageOptions().getService()) {
+      StorageClient storageClient = storage.storageClient;
+
+      SettableApiFuture<WriteObjectResponse> done = SettableApiFuture.create();
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      //noinspection resource
+      GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel channel =
+          new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      se.printStackTrace(System.out);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  private static @NonNull ResumableWrite getResumableWrite(String uploadId) {
+    StartResumableWriteRequest req = StartResumableWriteRequest.getDefaultInstance();
+    StartResumableWriteResponse resp =
+        StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
+    return new ResumableWrite(
+        req, resp, id -> WriteObjectRequest.newBuilder().setUploadId(id).build());
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index e1eb11c9..5efbae9f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -58,6 +58,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.function.Function;
 import java.util.stream.IntStream;
@@ -315,4 +316,12 @@ public final class TestUtils {
         .collect(ImmutableSet.toImmutableSet())
         .toArray(new Storage.BucketField[0]);
   }
+
+  public static <T> Optional<T> last(List<T> l) {
+    if (l.isEmpty()) {
+      return Optional.empty();
+    } else {
+      return Optional.of(l.get(l.size() - 1));
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcPlainRequestLoggingInterceptor.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcPlainRequestLoggingInterceptor.java
index 7116108d..611f7fd1 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcPlainRequestLoggingInterceptor.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/GrpcPlainRequestLoggingInterceptor.java
@@ -20,6 +20,8 @@ import com.google.api.gax.grpc.GrpcInterceptorProvider;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.MessageOrBuilder;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.ReadObjectResponse;
 import com.google.storage.v2.WriteObjectRequest;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
@@ -109,6 +111,10 @@ public final class GrpcPlainRequestLoggingInterceptor implements ClientIntercept
   static String fmtProto(@NonNull Object obj) {
     if (obj instanceof WriteObjectRequest) {
       return fmtProto((WriteObjectRequest) obj);
+    } else if (obj instanceof BidiWriteObjectRequest) {
+      return fmtProto((BidiWriteObjectRequest) obj);
+    } else if (obj instanceof ReadObjectResponse) {
+      return fmtProto((ReadObjectResponse) obj);
     } else if (obj instanceof MessageOrBuilder) {
       return fmtProto((MessageOrBuilder) obj);
     } else {
@@ -137,6 +143,38 @@ public final class GrpcPlainRequestLoggingInterceptor implements ClientIntercept
     return msg.toString();
   }
 
+  @NonNull
+  static String fmtProto(@NonNull BidiWriteObjectRequest msg) {
+    if (msg.hasChecksummedData()) {
+      ByteString content = msg.getChecksummedData().getContent();
+      if (content.size() > 20) {
+        BidiWriteObjectRequest.Builder b = msg.toBuilder();
+        ByteString snip = ByteString.copyFromUtf8(String.format("<snip (%d)>", content.size()));
+        ByteString trim = content.substring(0, 20).concat(snip);
+        b.getChecksummedDataBuilder().setContent(trim);
+
+        return b.build().toString();
+      }
+    }
+    return msg.toString();
+  }
+
+  @NonNull
+  static String fmtProto(@NonNull ReadObjectResponse msg) {
+    if (msg.hasChecksummedData()) {
+      ByteString content = msg.getChecksummedData().getContent();
+      if (content.size() > 20) {
+        ReadObjectResponse.Builder b = msg.toBuilder();
+        ByteString snip = ByteString.copyFromUtf8(String.format("<snip (%d)>", content.size()));
+        ByteString trim = content.substring(0, 20).concat(snip);
+        b.getChecksummedDataBuilder().setContent(trim);
+
+        return b.build().toString();
+      }
+    }
+    return msg.toString();
+  }
+
   private static final class InterceptorProvider implements GrpcInterceptorProvider {
     private static final InterceptorProvider INSTANCE = new InterceptorProvider();
 
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index 24ac6a0b..4d6f0aa0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage.it;
 
+import static com.google.cloud.storage.TestUtils.assertAll;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.gax.paging.Page;
@@ -38,6 +39,7 @@ import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.CreateHmacKeyOption;
 import com.google.cloud.storage.Storage.ListHmacKeysOption;
+import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
 import com.google.cloud.storage.it.runner.annotations.Backend;
@@ -246,4 +248,15 @@ public final class ITGrpcTest {
       s.delete(bucket.getName());
     }
   }
+
+  @Test
+  public void testGrpcUniverseDomainMatchesHost() throws Exception {
+    Storage storage =
+        StorageOptions.grpc().setUniverseDomain("my-universe-domain.com").build().getService();
+    assertAll(
+        () -> assertThat(storage.getOptions().getUniverseDomain().equals("my-universe-domain.com")),
+        () ->
+            assertThat(
+                storage.getOptions().getHost().equals("https://storage.my-universe-domain.com")));
+  }
 }