Repository: googleapis/java-storage
Pull request #2570 (2024-05-31 16:54:22)

fix: update grpc bidi resumable uploads to validate ack'd object size
Follow up to #2527 and #2514

The tests added in this PR account for ~940 of the added lines, however their scenarios are the same as the other upload scenario tests just using the different proto types. The remaining ~300 lines are the fix and small refactor to get the tests passing. I've left individual commits in the PR to make it easier to drill down if needed.


* Part 1, add tests and simulation server
  * Tests are a port of those added in #2527 using BidiWriteObjectRequest and BidiWriteObjectResponse
  * simulation server is a port of com.google.cloud.storage.ITGapicUnbufferedWritableByteChannelTest.DirectWriteService using BidiWriteObjectRequest and BidiWriteObjectResponse

* Cleanup unused inner-class

* Move request trimming inline rather than at the end

* Update GapicBidiUnbufferedWritableByteChannel to correctly handle failure scenarios {1, 2, 3, 4, 4_1, 4_2, 5, 7}
    
  Refactor ResumableSessionFailureScenario to accept Message and List rather than WriteObjectResponse and List<WriteObjectRequest>.
    
  The shape of BidiWriteObjectRequest/WriteObjectRequest and BidiWriteObjectResponse/WriteObjectResponse are largely the same (bidi has two extra fields) so rather than overloading toStorageException yet again, this time there is a single method for grpc messages that internally can branch when formatting the request. In this case, we're quite safe taking this relaxed typing because it is an internal API that is only called from a grpc context where protos will be used.

* Update GapicBidiUnbufferedWritableByteChannel to correctly handle partial consumption of content
    
  Tests passing now.
    
  Separate tracking of client detected errors and stream errors.
    
  When await is invoked, if a client detected error is present AND a stream error is present, the client detected error will take precedence. If a stream error happens and the client detected error has not yet been observed, the stream error will be added as a suppressed exception to the client detected error.


Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index 5db36a5f..359fe71c 100644
--- a/.github/.OwlBot.lock.yaml
+++ b/.github/.OwlBot.lock.yaml
@@ -13,5 +13,5 @@
 # limitations under the License.
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-java:latest
-  digest: sha256:68ba5f5164a4b55529d358bb262feaa000536a0c62980727dd05a91bbb47ea5e
-# created: 2024-05-09T16:31:37.168667071Z
+  digest: sha256:72f0d373307d128b2cb720c5cb4d90b31f0e86529dd138c632710ae0c69efae3
+# created: 2024-06-05T18:32:21.724930324Z
diff --git a/.github/workflows/unmanaged_dependency_check.yaml b/.github/workflows/unmanaged_dependency_check.yaml
index 50b31a20..0149c56d 100644
--- a/.github/workflows/unmanaged_dependency_check.yaml
+++ b/.github/workflows/unmanaged_dependency_check.yaml
@@ -17,6 +17,6 @@ jobs:
           # repository
           .kokoro/build.sh
       - name: Unmanaged dependency check
-        uses: googleapis/sdk-platform-java/java-shared-dependencies/unmanaged-dependency-check@google-cloud-shared-dependencies/v3.30.1
+        uses: googleapis/sdk-platform-java/java-shared-dependencies/unmanaged-dependency-check@google-cloud-shared-dependencies/v3.31.0
         with:
           bom-path: google-cloud-storage-bom/pom.xml
diff --git a/.kokoro/presubmit/graalvm-native-17.cfg b/.kokoro/presubmit/graalvm-native-17.cfg
index c283c89e..39531095 100644
--- a/.kokoro/presubmit/graalvm-native-17.cfg
+++ b/.kokoro/presubmit/graalvm-native-17.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.30.1"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.31.0"
 }
 
 env_vars: {
diff --git a/.kokoro/presubmit/graalvm-native.cfg b/.kokoro/presubmit/graalvm-native.cfg
index c997432f..c8748d5b 100644
--- a/.kokoro/presubmit/graalvm-native.cfg
+++ b/.kokoro/presubmit/graalvm-native.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.30.1"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.31.0"
 }
 
 env_vars: {
diff --git a/.readme-partials.yaml b/.readme-partials.yaml
index 2f8c71d3..cb154392 100644
--- a/.readme-partials.yaml
+++ b/.readme-partials.yaml
@@ -32,19 +32,19 @@ custom_content: |
   <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
   </dependency>
   ```
   
   If you are using Gradle 5.x or later, add this to your dependencies:
   ```Groovy
-  implementation platform('com.google.cloud:libraries-bom:2.39.0')
+  implementation platform('com.google.cloud:libraries-bom:2.40.0')
   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.39.0-beta' <!-- {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/CHANGELOG.md b/CHANGELOG.md
index 4d94b38a..33a2b4b9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,29 @@
 # Changelog
 
+## [2.40.0](https://github.com/googleapis/java-storage/compare/v2.39.0...v2.40.0) (2024-06-06)
+
+
+### Features
+
+* Promote google-cloud-storage-control to GA ([#2575](https://github.com/googleapis/java-storage/issues/2575)) ([129f188](https://github.com/googleapis/java-storage/commit/129f188a9877b3bd71db3b00ba3d8d6e4095fb67))
+
+
+### Bug Fixes
+
+* Reduce Java 21 Virtual Thread Pinning in IO operations ([#2553](https://github.com/googleapis/java-storage/issues/2553)) ([498fd0b](https://github.com/googleapis/java-storage/commit/498fd0b391c9d1d01c453fe31abea7890c0b498f))
+
+
+### Dependencies
+
+* Update dependency com.google.cloud:sdk-platform-java-config to v3.31.0 ([#2571](https://github.com/googleapis/java-storage/issues/2571)) ([67ce3d6](https://github.com/googleapis/java-storage/commit/67ce3d6505e1e48f496ab52884f5f5c2852aa445))
+* Update dependency net.jqwik:jqwik to v1.8.5 ([#2563](https://github.com/googleapis/java-storage/issues/2563)) ([88f7d86](https://github.com/googleapis/java-storage/commit/88f7d86d960aa6986cf41165d80d03a551502ec2))
+
+
+### Documentation
+
+* Managed Folders samples ([#2562](https://github.com/googleapis/java-storage/issues/2562)) ([5ffc1f2](https://github.com/googleapis/java-storage/commit/5ffc1f2fa75ac2e1c0022301e02b6ea6627faf03))
+* Update javadoc for createFrom ([#2522](https://github.com/googleapis/java-storage/issues/2522)) ([dc31e95](https://github.com/googleapis/java-storage/commit/dc31e95158e66d0eb4c85122edd29431a610a204))
+
 ## [2.39.0](https://github.com/googleapis/java-storage/compare/v2.38.0...v2.39.0) (2024-05-22)
 
 
diff --git a/README.md b/README.md
index 1a1f2b9d..2b7d1dc6 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.39.0</version>
+      <version>26.40.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -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.38.0</version>
+  <version>2.39.0</version>
 </dependency>
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-control</artifactId>
-  <version>2.39.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.39.0')
+implementation platform('com.google.cloud:libraries-bom:26.40.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.38.0'
+implementation 'com.google.cloud:google-cloud-storage:2.39.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.38.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.39.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -141,19 +141,19 @@ 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.39.0-beta</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.40.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
 </dependency>
 ```
 
 If you are using Gradle 5.x or later, add this to your dependencies:
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:2.39.0')
+implementation platform('com.google.cloud:libraries-bom:2.40.0')
 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.39.0-beta' <!-- {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
@@ -349,6 +349,10 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-storage/tree/
 | Delete Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/DeleteHmacKey.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/DeleteHmacKey.java) |
 | Get Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/GetHmacKey.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/GetHmacKey.java) |
 | List Hmac Keys | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/ListHmacKeys.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/ListHmacKeys.java) |
+| Create Managed Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/managedfolders/CreateManagedFolder.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/managedfolders/CreateManagedFolder.java) |
+| Delete Managed Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/managedfolders/DeleteManagedFolder.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/managedfolders/DeleteManagedFolder.java) |
+| Get Managed Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/managedfolders/GetManagedFolder.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/managedfolders/GetManagedFolder.java) |
+| List Managed Folders | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/managedfolders/ListManagedFolders.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/managedfolders/ListManagedFolders.java) |
 | Add File Owner | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/object/AddFileOwner.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/object/AddFileOwner.java) |
 | Batch Set Object Metadata | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/object/BatchSetObjectMetadata.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/object/BatchSetObjectMetadata.java) |
 | Change Object Csek To Kms | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/object/ChangeObjectCsekToKms.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/object/ChangeObjectCsekToKms.java) |
@@ -494,7 +498,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.38.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.39.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 fcc442cd..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.39.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.39.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 0c56765c..0af6b000 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -19,12 +19,12 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-bom</artifactId>
-  <version>2.39.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>
     <artifactId>sdk-platform-java-config</artifactId>
-    <version>3.30.1</version>
+    <version>3.31.0</version>
     <relativePath/>
   </parent>
 
@@ -69,37 +69,37 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.39.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.39.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.39.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.39.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.39.0-beta</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.39.0-beta</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.39.0-beta</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 f1f24952..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.39.0-beta</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.39.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 2723fedf..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.39.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,11 +12,11 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>2.39.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>
-    <pubsub-proto.version>1.111.5</pubsub-proto.version>
+    <pubsub-proto.version>1.112.0</pubsub-proto.version>
     <junit-platform.version>5.10.2</junit-platform.version>
   </properties>
   <dependencies>
@@ -181,13 +181,13 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.137.0</version>
+      <version>0.138.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-kms</artifactId>
-      <version>2.46.0</version>
+      <version>2.47.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
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/GapicUnbufferedDirectWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java
index 861e867d..5e67440a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java
@@ -20,12 +20,14 @@ 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.cloud.storage.WriteCtx.SimpleWriteObjectRequestBuilderFactory;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -34,11 +36,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 GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritableByteChannel {
@@ -55,13 +53,14 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
   private boolean open = true;
   private boolean first = true;
   private boolean finished = false;
+  private volatile WriteObjectRequest lastWrittenRequest;
 
   GapicUnbufferedDirectWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      SimpleWriteObjectRequestBuilderFactory requestFactory) {
-    String bucketName = requestFactory.bucketName();
+      WriteCtx<SimpleWriteObjectRequestBuilderFactory> writeCtx) {
+    String bucketName = writeCtx.getRequestFactory().bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
@@ -69,20 +68,63 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
         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
   public long write(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
-    return internalWrite(srcs, srcsOffset, srcsLength, false);
-  }
+    if (!open) {
+      throw new ClosedChannelException();
+    }
 
-  @Override
-  public long writeAndClose(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
-    long write = internalWrite(srcs, srcsOffset, srcsLength, true);
-    close();
-    return write;
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+    if (data.length == 0) {
+      return 0;
+    }
+
+    try {
+      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();
+        if (!first) {
+          builder.clearWriteObjectSpec();
+          builder.clearObjectChecksums();
+        }
+        builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
+        if (!datum.isOnlyFullBlocks()) {
+          builder.setFinishWrite(true);
+          if (cumulative != null) {
+            builder.setObjectChecksums(
+                ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+          }
+          finished = true;
+        }
+
+        WriteObjectRequest build = builder.build();
+        first = false;
+        bytesConsumed += contentSize;
+        lastWrittenRequest = build;
+        openedStream.onNext(build);
+      }
+      return bytesConsumed;
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
+    }
   }
 
   @Override
@@ -95,6 +137,7 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
     ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
     if (!finished) {
       WriteObjectRequest message = finishMessage();
+      lastWrittenRequest = message;
       try {
         openedStream.onNext(message);
         openedStream.onCompleted();
@@ -115,79 +158,22 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
     responseObserver.await();
   }
 
-  private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
-      throws ClosedChannelException {
-    if (!open) {
-      throw new ClosedChannelException();
-    }
-
-    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
-
-    List<WriteObjectRequest> messages = new ArrayList<>();
-
-    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());
-        }
-        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);
-      }
-    } catch (RuntimeException e) {
-      resultFuture.setException(e);
-      throw e;
-    }
-
-    return bytesConsumed;
-  }
-
   @NonNull
   private WriteObjectRequest finishMessage() {
     long offset = writeCtx.getTotalSentBytes().get();
     Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
 
-    WriteObjectRequest.Builder b =
-        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    WriteObjectRequest.Builder b = writeCtx.newRequestBuilder();
+    if (!first) {
+      b.clearWriteObjectSpec();
+      b.clearObjectChecksums();
+      first = false;
+    }
+    b.setFinishWrite(true).setWriteOffset(offset);
     if (crc32cValue != null) {
       b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
     }
-    WriteObjectRequest message = possiblyPairDownRequest(b, first).build();
-    return message;
+    return b.build();
   }
 
   private ApiStreamObserver<WriteObjectRequest> openedStream() {
@@ -201,48 +187,20 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
     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;
     }
 
@@ -257,15 +215,58 @@ final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritab
      */
     @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(), getRequests(), null, context, t);
+        invocationHandle.setException(storageException);
+      } else {
+        invocationHandle.setException(t);
+      }
     }
 
     @Override
     public void onCompleted() {
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
+      try {
+        if (last == null) {
+          throw new StorageException(
+              0, "onComplete without preceding onNext, unable to determine success.");
+        } else if (last.hasResource()) {
+          long totalSentBytes = writeCtx.getTotalSentBytes().get();
+          long finalSize = last.getResource().getSize();
+          if (totalSentBytes == finalSize) {
+            writeCtx.getConfirmedBytes().set(finalSize);
+            resultFuture.set(last);
+          } else if (finalSize < totalSentBytes) {
+            throw ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                getRequests(), last, context, null);
+          } else {
+            throw ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                getRequests(), last, context, null);
+          }
+        } else {
+          throw ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+              getRequests(), last, context, null);
+        }
+      } catch (Throwable se) {
+        open = false;
+        invocationHandle.setException(se);
+      } finally {
+        invocationHandle.set(null);
+      }
+    }
+
+    private @NonNull ImmutableList<@NonNull WriteObjectRequest> getRequests() {
+      if (lastWrittenRequest == null) {
+        return ImmutableList.of();
+      } else {
+        return ImmutableList.of(lastWrittenRequest);
       }
-      invocationHandle.set(null);
     }
 
     void await() {
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 88540533..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
@@ -185,7 +185,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         resultFuture,
                         getChunkSegmenter(),
                         write,
-                        WriteObjectRequestBuilderFactory.simple(start)))
+                        new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(start))))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
     }
@@ -213,7 +213,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         resultFuture,
                         getChunkSegmenter(),
                         write,
-                        WriteObjectRequestBuilderFactory.simple(start)))
+                        new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(start))))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
@@ -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/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 294d481c..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;
@@ -64,11 +64,11 @@ enum ResumableSessionFailureScenario {
   SCENARIO_4_1(
       BaseServiceException.UNKNOWN_CODE,
       "dataLoss",
-      "Finalized resumable session, but object size less than expected."),
+      "Finalized upload, but object size less than expected."),
   SCENARIO_4_2(
       BaseServiceException.UNKNOWN_CODE,
       "dataLoss",
-      "Finalized resumable session, but object size greater than expected."),
+      "Finalized upload, but object size greater than expected."),
   SCENARIO_5(
       BaseServiceException.UNKNOWN_CODE,
       "dataLoss",
@@ -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/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 1a860a05..fca1d129 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -2906,18 +2906,16 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
   Blob createFrom(BlobInfo blobInfo, Path path, BlobWriteOption... options) throws IOException;
 
   /**
-   * Uploads {@code path} to the blob using {@link #writer} and {@code bufferSize}. By default any
-   * MD5 and CRC32C values in the given {@code blobInfo} are ignored unless requested via the {@link
+   * Uploads {@code path} to the blob using {@code ResumableSession}. By default any MD5 and CRC32C
+   * values in the given {@code blobInfo} are ignored unless requested via the {@link
    * BlobWriteOption#md5Match()} and {@link BlobWriteOption#crc32cMatch()} options. Folder upload is
    * not supported. Note that all <a href="https://cloud.google.com/storage/docs/metadata#fixed">
    * non-editable metadata</a>, such as generation or metageneration, will be ignored even if it's
    * present in the provided BlobInfo object.
    *
-   * <p>{@link #createFrom(BlobInfo, Path, BlobWriteOption...)} invokes this method with a buffer
-   * size of 15 MiB. Users can pass alternative values. Larger buffer sizes might improve the upload
-   * performance but require more memory. This can cause an OutOfMemoryError or add significant
-   * garbage collection overhead. Smaller buffer sizes reduce memory consumption, that is noticeable
-   * when uploading many objects in parallel. Buffer sizes less than 256 KiB are treated as 256 KiB.
+   * <p>This method used to preallocate a buffer, but since v2.25.0, it uses a ResumableSession and
+   * no longer needs it. The bufferSize parameter is still present for binary compatibility, but is
+   * now ignored.
    *
    * <p>Example of uploading a humongous file:
    *
@@ -2925,14 +2923,13 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * BlobId blobId = BlobId.of(bucketName, blobName);
    * BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("video/webm").build();
    *
-   * int largeBufferSize = 150 * 1024 * 1024;
    * Path file = Paths.get("humongous.file");
-   * storage.createFrom(blobInfo, file, largeBufferSize);
+   * storage.createFrom(blobInfo, file, 0);
    * }</pre>
    *
    * @param blobInfo blob to create
    * @param path file to upload
-   * @param bufferSize size of the buffer I/O operations
+   * @param bufferSize ignored field, still present for compatibility purposes
    * @param options blob write options
    * @return a {@code Blob} with complete information
    * @throws IOException on I/O error
diff --git a/grpc-google-cloud-storage-control-v2/pom.xml b/grpc-google-cloud-storage-control-v2/pom.xml
index dd513cc3..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.39.0-beta</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.39.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 4f9b97ee..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.39.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.39.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 10bb5f3f..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.39.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>
@@ -14,7 +14,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>sdk-platform-java-config</artifactId>
-    <version>3.30.1</version>
+    <version>3.31.0</version>
   </parent>
 
   <developers>
@@ -76,17 +76,17 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage</artifactId>
-        <version>2.39.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>
         <artifactId>google-cloud-pubsub</artifactId>
-        <version>1.129.5</version>
+        <version>1.130.0</version>
         <scope>test</scope>
       </dependency>
       <dependency>
@@ -98,32 +98,32 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-v2</artifactId>
-        <version>2.39.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.39.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.39.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.39.0-beta</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.39.0-beta</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.39.0-beta</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 fa454bb1..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.39.0-beta</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.39.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 dfc85cdc..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.39.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.39.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 d4893b17..1388f8c4 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.38.0</version>
+      <version>2.39.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.39.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] -->
 
@@ -66,7 +66,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.5</version>
+      <version>1.130.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index 9229c8e1..2e9a84a9 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -29,7 +29,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.39.0</version>
+        <version>26.40.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -61,13 +61,13 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.5</version>
+      <version>1.130.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.39.0-beta</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/pom.xml b/samples/pom.xml
index c1b0d76c..7c252467 100644
--- a/samples/pom.xml
+++ b/samples/pom.xml
@@ -48,7 +48,7 @@
       <plugin>
         <groupId>org.sonatype.plugins</groupId>
         <artifactId>nexus-staging-maven-plugin</artifactId>
-        <version>1.6.13</version>
+        <version>1.7.0</version>
         <configuration>
           <skipNexusStagingDeployMojo>true</skipNexusStagingDeployMojo>
         </configuration>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 2012436e..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.39.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.39.0-beta</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>
 
@@ -58,7 +58,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.5</version>
+      <version>1.130.0</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 7be5d0f9..8941aebf 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.39.0</version>
+        <version>26.40.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -76,7 +76,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.5</version>
+      <version>1.130.0</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java b/samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java
new file mode 100644
index 00000000..6bfc6609
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java
@@ -0,0 +1,50 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_control_create_folder]
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.CreateFolderRequest;
+import com.google.storage.control.v2.Folder;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+
+public final class CreateFolder {
+
+  public static void createFolder(String bucketName, String folderName) throws IOException {
+    // The name of the bucket
+    // String bucketName = "your-unique-bucket-name";
+
+    // The name of the folder within the bucket
+    // String folderName = "your-unique-folder-name";
+
+    try (StorageControlClient storageControl = StorageControlClient.create()) {
+
+      CreateFolderRequest request =
+          CreateFolderRequest.newBuilder()
+              // Set project to "_" to signify globally scoped bucket
+              .setParent(BucketName.format("_", bucketName))
+              .setFolderId(folderName)
+              .build();
+
+      Folder newFolder = storageControl.createFolder(request);
+
+      System.out.printf("Created folder: %s%n", newFolder.getName());
+    }
+  }
+}
+// [END storage_control_create_folder]
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java b/samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java
new file mode 100644
index 00000000..f6d0868c
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java
@@ -0,0 +1,54 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_create_bucket_hierarchical_namespace]
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.HierarchicalNamespace;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+
+public final class CreateHierarchicalNamespaceBucket {
+
+  public static void createHierarchicalNamespaceBucket(String projectId, String bucketName)
+      throws Exception {
+    // The ID of your GCP project
+    // String projectId = "your-project-id";
+
+    // The ID to give your GCS bucket
+    // String bucketName = "your-unique-bucket-name";
+    StorageOptions storageOptions = StorageOptions.newBuilder().setProjectId(projectId).build();
+    try (Storage storage = storageOptions.getService()) {
+
+      BucketInfo bucketInfo =
+          BucketInfo.newBuilder(bucketName)
+              .setIamConfiguration(
+                  // Hierarchical namespace buckets must use uniform bucket-level access.
+                  IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+              .setHierarchicalNamespace(HierarchicalNamespace.newBuilder().setEnabled(true).build())
+              .build();
+
+      Bucket bucket = storage.create(bucketInfo);
+
+      System.out.printf(
+          "Created bucket %s with Hierarchical Namespace enabled.%n", bucket.getName());
+    }
+  }
+}
+// [END storage_create_bucket_hierarchical_namespace]
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java b/samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java
new file mode 100644
index 00000000..187a682f
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java
@@ -0,0 +1,48 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_control_delete_folder]
+
+import com.google.storage.control.v2.DeleteFolderRequest;
+import com.google.storage.control.v2.FolderName;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+
+public final class DeleteFolder {
+
+  public static void deleteFolder(String bucketName, String folderName) throws IOException {
+    // The name of the bucket
+    // String bucketName = "your-unique-bucket-name";
+
+    // The name of the folder within the bucket
+    // String folderName = "your-unique-folder-name";
+
+    try (StorageControlClient storageControl = StorageControlClient.create()) {
+
+      // Set project to "_" to signify globally scoped bucket
+      String folderResourceName = FolderName.format("_", bucketName, folderName);
+      DeleteFolderRequest request =
+          DeleteFolderRequest.newBuilder().setName(folderResourceName).build();
+
+      storageControl.deleteFolder(request);
+
+      System.out.printf("Deleted folder: %s%n", folderResourceName);
+    }
+  }
+}
+// [END storage_control_delete_folder]
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java b/samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java
new file mode 100644
index 00000000..454ed7b3
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java
@@ -0,0 +1,50 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_control_get_folder]
+
+import com.google.storage.control.v2.Folder;
+import com.google.storage.control.v2.FolderName;
+import com.google.storage.control.v2.GetFolderRequest;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+
+public final class GetFolder {
+
+  public static void getFolder(String bucketName, String folderName) throws IOException {
+    // The name of the bucket
+    // String bucketName = "your-unique-bucket-name";
+
+    // The name of the folder within the bucket
+    // String folderName = "your-unique-folder-name";
+
+    try (StorageControlClient storageControl = StorageControlClient.create()) {
+
+      GetFolderRequest request =
+          GetFolderRequest.newBuilder()
+              // Set project to "_" to signify globally scoped bucket
+              .setName(FolderName.format("_", bucketName, folderName))
+              .build();
+
+      Folder newFolder = storageControl.getFolder(request);
+
+      System.out.printf("Got folder: %s%n", newFolder.getName());
+    }
+  }
+}
+// [END storage_control_get_folder]
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java b/samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java
new file mode 100644
index 00000000..2778213e
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java
@@ -0,0 +1,48 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_control_list_folders]
+
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.Folder;
+import com.google.storage.control.v2.ListFoldersRequest;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+
+public final class ListFolders {
+
+  public static void listFolders(String bucketName) throws IOException {
+    // The name of the bucket
+    // String bucketName = "your-unique-bucket-name";
+
+    try (StorageControlClient storageControl = StorageControlClient.create()) {
+
+      ListFoldersRequest request =
+          ListFoldersRequest.newBuilder()
+              // Set project to "_" to signify globally scoped bucket
+              .setParent(BucketName.format("_", bucketName))
+              .build();
+
+      Iterable<Folder> folders = storageControl.listFolders(request).iterateAll();
+      for (Folder folder : folders) {
+        System.out.printf("Found folder: %s%n", folder.getName());
+      }
+    }
+  }
+}
+// [END storage_control_list_folders]
diff --git a/samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java b/samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java
new file mode 100644
index 00000000..e8f95b6f
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java
@@ -0,0 +1,66 @@
+/*
+ * 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.example.storage.control.v2;
+
+// [START storage_control_rename_folder]
+
+import com.google.api.gax.longrunning.OperationFuture;
+import com.google.storage.control.v2.Folder;
+import com.google.storage.control.v2.FolderName;
+import com.google.storage.control.v2.RenameFolderMetadata;
+import com.google.storage.control.v2.RenameFolderRequest;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public final class RenameFolder {
+
+  public static void renameFolder(
+      String bucketName, String sourceFolderName, String destinationFolderName)
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    // The name of the bucket
+    // String bucketName = "your-unique-bucket-name";
+
+    // The name of the folder within the bucket
+    // String sourceFolderName = "your-unique-source-folder-name";
+
+    // The new name of the folder within the bucket
+    // String destinationFolderName = "your-unique-destination-folder-name";
+
+    try (StorageControlClient storageControl = StorageControlClient.create()) {
+
+      // Set project to "_" to signify globally scoped bucket
+      String sourceFolderResourceName = FolderName.format("_", bucketName, sourceFolderName);
+      RenameFolderRequest request =
+          RenameFolderRequest.newBuilder()
+              .setName(sourceFolderResourceName)
+              .setDestinationFolderId(destinationFolderName)
+              .build();
+
+      OperationFuture<Folder, RenameFolderMetadata> renameOperation =
+          storageControl.renameFolderAsync(request);
+
+      Folder destinationFolder = renameOperation.get(30, TimeUnit.SECONDS);
+
+      System.out.printf(
+          "Renamed folder from %s to %s%n", sourceFolderResourceName, destinationFolder.getName());
+    }
+  }
+}
+// [END storage_control_rename_folder]
diff --git a/samples/snippets/src/main/java/com/example/storage/managedfolders/CreateManagedFolder.java b/samples/snippets/src/main/java/com/example/storage/managedfolders/CreateManagedFolder.java
new file mode 100644
index 00000000..23744ed6
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/managedfolders/CreateManagedFolder.java
@@ -0,0 +1,44 @@
+/*
+ * 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.example.storage.managedfolders;
+
+// [START storage_control_managed_folder_create]
+
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.CreateManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.StorageControlClient;
+
+public class CreateManagedFolder {
+  public static void managedFolderCreate(String bucketName, String managedFolderId)
+      throws Exception {
+
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (StorageControlClient storageControlClient = StorageControlClient.create()) {
+      CreateManagedFolderRequest request =
+          CreateManagedFolderRequest.newBuilder()
+              // Set project to "_" to signify global bucket
+              .setParent(BucketName.format("_", bucketName))
+              .setManagedFolder(ManagedFolder.newBuilder().build())
+              .setManagedFolderId(managedFolderId)
+              .build();
+      String response = storageControlClient.createManagedFolder(request).getName();
+      System.out.printf("Performed createManagedFolder request for %s%n", response);
+    }
+  }
+}
+// [END storage_control_managed_folder_create]
diff --git a/samples/snippets/src/main/java/com/example/storage/managedfolders/DeleteManagedFolder.java b/samples/snippets/src/main/java/com/example/storage/managedfolders/DeleteManagedFolder.java
new file mode 100644
index 00000000..b72640c1
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/managedfolders/DeleteManagedFolder.java
@@ -0,0 +1,46 @@
+/*
+ * 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.example.storage.managedfolders;
+
+// [START storage_control_managed_folder_delete]
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.DeleteManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolderName;
+import com.google.storage.control.v2.StorageControlClient;
+
+class DeleteManagedFolder {
+  public static void managedFolderDelete(String bucketName, String managedFolderId)
+      throws Exception {
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (StorageControlClient storageControlClient = StorageControlClient.create()) {
+      // Set project to "_" to signify global bucket
+      BucketName resourceBucketName = BucketName.of("_", bucketName);
+      DeleteManagedFolderRequest deleteManagedFolderRequest =
+          DeleteManagedFolderRequest.newBuilder()
+              .setName(
+                  ManagedFolderName.format(
+                      resourceBucketName.getProject(),
+                      resourceBucketName.getBucket(),
+                      managedFolderId))
+              .build();
+      storageControlClient.deleteManagedFolder(deleteManagedFolderRequest);
+      System.out.printf("Deleted Managed Folder %s%n", managedFolderId);
+    }
+  }
+}
+
+// [END storage_control_managed_folder_delete]
diff --git a/samples/snippets/src/main/java/com/example/storage/managedfolders/GetManagedFolder.java b/samples/snippets/src/main/java/com/example/storage/managedfolders/GetManagedFolder.java
new file mode 100644
index 00000000..683444c1
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/managedfolders/GetManagedFolder.java
@@ -0,0 +1,48 @@
+/*
+ * 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.example.storage.managedfolders;
+
+// [START storage_control_managed_folder_get]
+
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.GetManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.ManagedFolderName;
+import com.google.storage.control.v2.StorageControlClient;
+
+class GetManagedFolder {
+
+  public static void managedFolderGet(String bucketName, String managedFolderId) throws Exception {
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (StorageControlClient storageControlClient = StorageControlClient.create()) {
+      // Set project to "_" to signify global bucket
+      BucketName resourceBucketName = BucketName.of("_", bucketName);
+      GetManagedFolderRequest getManagedFolderRequest =
+          GetManagedFolderRequest.newBuilder()
+              .setName(
+                  ManagedFolderName.format(
+                      resourceBucketName.getProject(),
+                      resourceBucketName.getBucket(),
+                      managedFolderId))
+              .build();
+      ManagedFolder managedFolder = storageControlClient.getManagedFolder(getManagedFolderRequest);
+      System.out.printf("Got Managed Folder %s%n", managedFolder.getName());
+    }
+  }
+}
+
+// [END storage_control_managed_folder_get]
diff --git a/samples/snippets/src/main/java/com/example/storage/managedfolders/ListManagedFolders.java b/samples/snippets/src/main/java/com/example/storage/managedfolders/ListManagedFolders.java
new file mode 100644
index 00000000..bb7ce977
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/managedfolders/ListManagedFolders.java
@@ -0,0 +1,45 @@
+/*
+ * 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.example.storage.managedfolders;
+
+// [START storage_control_managed_folder_list]
+
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.ListManagedFoldersRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.StorageControlClient;
+
+class ListManagedFolders {
+
+  public static void managedFolderList(String bucketName) throws Exception {
+    // Instantiates a client in a try-with-resource to automatically cleanup underlying resources
+    try (StorageControlClient storageControlClient = StorageControlClient.create()) {
+      ListManagedFoldersRequest listManagedFoldersRequest =
+          ListManagedFoldersRequest.newBuilder()
+              // Set project to "_" to signify global bucket
+              .setParent(BucketName.format("_", bucketName))
+              .build();
+      Iterable<ManagedFolder> managedFolders =
+          storageControlClient.listManagedFolders(listManagedFoldersRequest).iterateAll();
+      for (ManagedFolder folder : managedFolders) {
+        System.out.printf("%s bucket has managed folder %s%n", bucketName, folder.getName());
+      }
+    }
+  }
+}
+
+// [END storage_control_managed_folder_list]
diff --git a/storage-shared-benchmarking/pom.xml b/storage-shared-benchmarking/pom.xml
index ba49f60b..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.39.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.39.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 46958f51..39d03adf 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,10 +1,10 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:2.39.0:2.39.0
-gapic-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
-grpc-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
-proto-google-cloud-storage-v2:2.39.0-alpha:2.39.0-alpha
-google-cloud-storage-control:2.39.0-beta:2.39.0-beta
-proto-google-cloud-storage-control-v2:2.39.0-beta:2.39.0-beta
-grpc-google-cloud-storage-control-v2:2.39.0-beta:2.39.0-beta
+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/main/java/com/google/cloud/storage/testing/RemoteStorageHelper.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/RemoteStorageHelper.java
index 7d4dc81d..50bfdc87 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/RemoteStorageHelper.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/RemoteStorageHelper.java
@@ -99,7 +99,7 @@ public class RemoteStorageHelper {
                   }
                   forceDelete(storage, bucket.getName());
                 } catch (Exception e) {
-                  // Ignore the exception, maybe the bucket is being deleted by someone else.
+                  log.info("Failed to clean buckets " + e.getMessage());
                 }
               }
             }
@@ -298,6 +298,7 @@ public class RemoteStorageHelper {
           }
           return true;
         } catch (StorageException e) {
+          log.warning("Caught exception in Delete Bucket Task" + e.getMessage());
           if (e.getCode() == 409) {
             try {
               Thread.sleep(500);
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/ITGapicUnbufferedDirectWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedDirectWritableByteChannelTest.java
new file mode 100644
index 00000000..4a70e9a2
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedDirectWritableByteChannelTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.cloud.storage.WriteCtx.SimpleWriteObjectRequestBuilderFactory;
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.storage.v2.Object;
+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.junit.Test;
+
+public final class ITGapicUnbufferedDirectWritableByteChannelTest {
+
+  private static final ChunkSegmenter CHUNK_SEGMENTER =
+      new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), _256KiB, _256KiB);
+
+  /** Attempting to finalize, ack equals expected */
+  @Test
+  public void ack_eq() throws Exception {
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder().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();
+      WriteCtx<SimpleWriteObjectRequestBuilderFactory> writeCtx =
+          new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(req1));
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      GapicUnbufferedDirectWritableByteChannel channel =
+          new GapicUnbufferedDirectWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      channel.close();
+
+      WriteObjectResponse writeObjectResponse = done.get(2, TimeUnit.SECONDS);
+      assertAll(
+          () -> assertThat(writeObjectResponse).isEqualTo(resp1),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(_256KiB),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /** Attempting to finalize, ack &lt; expected */
+  @Test
+  public void ack_lt() throws Exception {
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder().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();
+      WriteCtx<SimpleWriteObjectRequestBuilderFactory> writeCtx =
+          new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(req1));
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      //noinspection resource
+      GapicUnbufferedDirectWritableByteChannel channel =
+          new GapicUnbufferedDirectWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+
+  /** Attempting to finalize, ack &gt; expected */
+  @Test
+  public void ack_gt() throws Exception {
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder().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();
+      WriteCtx<SimpleWriteObjectRequestBuilderFactory> writeCtx =
+          new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(req1));
+      writeCtx.getTotalSentBytes().set(_512KiB);
+      writeCtx.getConfirmedBytes().set(0);
+
+      //noinspection resource
+      GapicUnbufferedDirectWritableByteChannel channel =
+          new GapicUnbufferedDirectWritableByteChannel(
+              done, CHUNK_SEGMENTER, storageClient.writeObjectCallable(), writeCtx);
+
+      StorageException se = assertThrows(StorageException.class, channel::close);
+      assertAll(
+          () -> assertThat(se.getCode()).isEqualTo(0),
+          () -> assertThat(se.getReason()).isEqualTo("dataLoss"),
+          () -> assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(0),
+          () -> assertThat(channel.isOpen()).isFalse());
+    }
+  }
+}
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/ITGapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
index 239b9bf3..326cbb15 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
@@ -153,11 +153,14 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
         new DirectWriteService(
             ImmutableMap.of(ImmutableList.of(req1, req2, req3, req4, req5), resp));
     try (FakeServer fake = FakeServer.of(service);
-        StorageClient sc = StorageClient.create(fake.storageSettings())) {
+        StorageClient sc =
+            PackagePrivateMethodWorkarounds.maybeGetStorageClient(
+                fake.getGrpcStorageOptions().getService())) {
+      assertThat(sc).isNotNull();
       SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
       try (GapicUnbufferedDirectWritableByteChannel c =
           new GapicUnbufferedDirectWritableByteChannel(
-              result, segmenter, sc.writeObjectCallable(), reqFactory)) {
+              result, segmenter, sc.writeObjectCallable(), new WriteCtx<>(reqFactory))) {
         c.write(ByteBuffer.wrap(bytes));
       }
       assertThat(result.get()).isEqualTo(resp);
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/runner/registry/BackendResources.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
index 8aab9ba6..4ed6887f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java
@@ -23,6 +23,8 @@ import static com.google.cloud.storage.it.runner.registry.RegistryApplicabilityP
 
 import com.google.cloud.NoCredentials;
 import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.GrpcStorageOptions;
+import com.google.cloud.storage.HttpStorageOptions;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.TransportCompatibility.Transport;
@@ -43,36 +45,10 @@ final class BackendResources implements ManagedLifecycle {
   private BackendResources(
       Backend backend,
       ProtectedBucketNames protectedBucketNames,
-      TestRunScopedInstance<StorageInstance> storageJson,
-      TestRunScopedInstance<StorageInstance> storageGrpc,
-      TestRunScopedInstance<BucketInfoShim> bucket,
-      TestRunScopedInstance<BucketInfoShim> bucketRequesterPays,
-      TestRunScopedInstance<ObjectsFixture> objectsFixture,
-      TestRunScopedInstance<ObjectsFixture> objectsFixtureRequesterPays,
-      TestRunScopedInstance<KmsFixture> kmsFixture) {
+      ImmutableList<RegistryEntry<?>> registryEntries) {
     this.backend = backend;
     this.protectedBucketNames = protectedBucketNames;
-    this.registryEntries =
-        ImmutableList.of(
-            RegistryEntry.of(
-                4, Storage.class, storageJson, transportAndBackendAre(Transport.HTTP, backend)),
-            RegistryEntry.of(
-                5, Storage.class, storageGrpc, transportAndBackendAre(Transport.GRPC, backend)),
-            RegistryEntry.of(
-                6,
-                BucketInfo.class,
-                bucketRequesterPays,
-                backendIs(backend).and(isRequesterPaysBucket())),
-            RegistryEntry.of(
-                7, BucketInfo.class, bucket, backendIs(backend).and(isDefaultBucket())),
-            RegistryEntry.of(
-                8, ObjectsFixture.class, objectsFixture, backendIs(backend).and(isDefaultBucket())),
-            RegistryEntry.of(
-                9,
-                ObjectsFixture.class,
-                objectsFixtureRequesterPays,
-                backendIs(backend).and(isRequesterPaysBucket())),
-            RegistryEntry.of(10, KmsFixture.class, kmsFixture, backendIs(backend)));
+    this.registryEntries = registryEntries;
   }
 
   public ImmutableList<RegistryEntry<?>> getRegistryEntries() {
@@ -97,41 +73,53 @@ final class BackendResources implements ManagedLifecycle {
     return MoreObjects.toStringHelper(this).add("backend", backend).toString();
   }
 
+  @SuppressWarnings("SwitchStatementWithTooFewBranches")
   static BackendResources of(Backend backend) {
     ProtectedBucketNames protectedBucketNames = new ProtectedBucketNames();
     TestRunScopedInstance<StorageInstance> storageJson =
         TestRunScopedInstance.of(
             "STORAGE_JSON_" + backend.name(),
-            () ->
-                new StorageInstance(
-                    backend == Backend.TEST_BENCH
-                        ? StorageOptions.http()
-                            .setCredentials(NoCredentials.getInstance())
-                            // TODO: improve this
-                            .setHost(Registry.getInstance().testBench().getBaseUri())
-                            .setProjectId("test-project-id")
-                            .build()
-                        : StorageOptions.http().build(),
-                    protectedBucketNames));
+            () -> {
+              HttpStorageOptions.Builder optionsBuilder;
+              switch (backend) {
+                case TEST_BENCH:
+                  optionsBuilder =
+                      StorageOptions.http()
+                          .setCredentials(NoCredentials.getInstance())
+                          .setHost(Registry.getInstance().testBench().getBaseUri())
+                          .setProjectId("test-project-id");
+                  break;
+                default: // PROD, java8 doesn't have exhaustive checking for enum switch
+                  optionsBuilder = StorageOptions.http();
+                  break;
+              }
+              HttpStorageOptions built = optionsBuilder.build();
+              return new StorageInstance(built, protectedBucketNames);
+            });
     TestRunScopedInstance<StorageInstance> storageGrpc =
         TestRunScopedInstance.of(
             "STORAGE_GRPC_" + backend.name(),
-            () ->
-                new StorageInstance(
-                    backend == Backend.TEST_BENCH
-                        ? StorageOptions.grpc()
-                            .setGrpcInterceptorProvider(
-                                GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
-                            .setCredentials(NoCredentials.getInstance())
-                            // TODO: improve this
-                            .setHost(Registry.getInstance().testBench().getGRPCBaseUri())
-                            .setProjectId("test-project-id")
-                            .build()
-                        : StorageOptions.grpc()
-                            .setGrpcInterceptorProvider(
-                                GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
-                            .build(),
-                    protectedBucketNames));
+            () -> {
+              GrpcStorageOptions.Builder optionsBuilder;
+              switch (backend) {
+                case TEST_BENCH:
+                  optionsBuilder =
+                      StorageOptions.grpc()
+                          .setCredentials(NoCredentials.getInstance())
+                          .setHost(Registry.getInstance().testBench().getGRPCBaseUri())
+                          .setProjectId("test-project-id");
+                  break;
+                default: // PROD, java8 doesn't have exhaustive checking for enum switch
+                  optionsBuilder = StorageOptions.grpc();
+                  break;
+              }
+              GrpcStorageOptions built =
+                  optionsBuilder
+                      .setGrpcInterceptorProvider(
+                          GrpcPlainRequestLoggingInterceptor.getInterceptorProvider())
+                      .build();
+              return new StorageInstance(built, protectedBucketNames);
+            });
     TestRunScopedInstance<BucketInfoShim> bucket =
         TestRunScopedInstance.of(
             "BUCKET_" + backend.name(),
@@ -154,7 +142,7 @@ final class BackendResources implements ManagedLifecycle {
         TestRunScopedInstance.of(
             "OBJECTS_FIXTURE_" + backend.name(),
             () -> new ObjectsFixture(storageJson.get().getStorage(), bucket.get().getBucketInfo()));
-    TestRunScopedInstance<ObjectsFixture> objectsFixtureRequesterPays =
+    TestRunScopedInstance<ObjectsFixture> objectsFixtureRp =
         TestRunScopedInstance.of(
             "OBJECTS_FIXTURE_REQUESTER_PAYS_" + backend.name(),
             () ->
@@ -166,12 +154,25 @@ final class BackendResources implements ManagedLifecycle {
     return new BackendResources(
         backend,
         protectedBucketNames,
-        storageJson,
-        storageGrpc,
-        bucket,
-        bucketRp,
-        objectsFixture,
-        objectsFixtureRequesterPays,
-        kmsFixture);
+        ImmutableList.of(
+            RegistryEntry.of(
+                40, Storage.class, storageJson, transportAndBackendAre(Transport.HTTP, backend)),
+            RegistryEntry.of(
+                50, Storage.class, storageGrpc, transportAndBackendAre(Transport.GRPC, backend)),
+            RegistryEntry.of(
+                60, BucketInfo.class, bucketRp, backendIs(backend).and(isRequesterPaysBucket())),
+            RegistryEntry.of(
+                70, BucketInfo.class, bucket, backendIs(backend).and(isDefaultBucket())),
+            RegistryEntry.of(
+                80,
+                ObjectsFixture.class,
+                objectsFixture,
+                backendIs(backend).and(isDefaultBucket())),
+            RegistryEntry.of(
+                90,
+                ObjectsFixture.class,
+                objectsFixtureRp,
+                backendIs(backend).and(isRequesterPaysBucket())),
+            RegistryEntry.of(100, KmsFixture.class, kmsFixture, backendIs(backend))));
   }
 }
diff --git a/samples/snippets/src/test/java/com/example/storage/ITStorageSnippets.java b/samples/snippets/src/test/java/com/example/storage/ITStorageSnippets.java
index 3c448def..9a33bbcd 100644
--- a/samples/snippets/src/test/java/com/example/storage/ITStorageSnippets.java
+++ b/samples/snippets/src/test/java/com/example/storage/ITStorageSnippets.java
@@ -67,12 +67,6 @@ public class ITStorageSnippets {
   @AfterClass
   public static void afterClass() throws ExecutionException, InterruptedException {
     if (storage != null) {
-      // In beforeClass, we make buckets auto-delete blobs older than a day old.
-      // Here, delete all buckets older than 2 days. They should already be empty and easy.
-      long cleanTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
-      long cleanTimeout = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(1);
-      RemoteStorageHelper.cleanBuckets(storage, cleanTime, cleanTimeout);
-
       boolean wasDeleted = RemoteStorageHelper.forceDelete(storage, BUCKET, 1, TimeUnit.MINUTES);
       if (!wasDeleted && log.isLoggable(Level.WARNING)) {
         log.log(Level.WARNING, "Deletion of bucket {0} timed out, bucket is not empty", BUCKET);
diff --git a/samples/snippets/src/test/java/com/example/storage/control/v2/FoldersTest.java b/samples/snippets/src/test/java/com/example/storage/control/v2/FoldersTest.java
new file mode 100644
index 00000000..7b4f8f21
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/control/v2/FoldersTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.example.storage.control.v2;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.gax.rpc.NotFoundException;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.HierarchicalNamespace;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.cloud.testing.junit4.StdOutCaptureRule;
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.Folder;
+import com.google.storage.control.v2.FolderName;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+public final class FoldersTest {
+
+  @Rule public StdOutCaptureRule stdOut = new StdOutCaptureRule();
+
+  private Storage storage;
+  private BucketInfo bucket;
+  private StorageControlClient storageControl;
+
+  @Before
+  public void setUp() throws Exception {
+    storage = StorageOptions.http().build().getService();
+    String bucketName = RemoteStorageHelper.generateBucketName();
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName)
+            .setIamConfiguration(
+                // Hierarchical namespace buckets must use uniform bucket-level access.
+                IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+            .setHierarchicalNamespace(HierarchicalNamespace.newBuilder().setEnabled(true).build())
+            .build();
+
+    bucket = storage.create(bucketInfo).asBucketInfo();
+    storageControl = StorageControlClient.create();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    // Use try-with-resource to handle the dance closing multiple things
+    //noinspection EmptyTryBlock
+    try (AutoCloseable ignore1 = storage;
+        AutoCloseable ignore2 = storageControl) {
+      // ignore
+    }
+  }
+
+  @Test
+  public void createFolder() throws IOException {
+    String folderName = UUID.randomUUID().toString();
+    CreateFolder.createFolder(bucket.getName(), folderName);
+    try {
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(folderName);
+    } finally {
+      storageControl.deleteFolder(FolderName.of("_", bucket.getName(), folderName));
+    }
+  }
+
+  @Test
+  public void getFolder() throws IOException {
+    FolderName folderName = FolderName.of("_", bucket.getName(), UUID.randomUUID().toString());
+    Folder gen1 =
+        storageControl.createFolder(
+            BucketName.of("_", bucket.getName()),
+            Folder.getDefaultInstance(),
+            folderName.getFolder());
+
+    GetFolder.getFolder(bucket.getName(), folderName.getFolder());
+    try {
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(folderName.toString());
+    } finally {
+      storageControl.deleteFolder(folderName);
+    }
+  }
+
+  @Test
+  public void renameFolder()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    FolderName srcFolderName = FolderName.of("_", bucket.getName(), UUID.randomUUID().toString());
+    Folder gen1 =
+        storageControl.createFolder(
+            BucketName.of("_", bucket.getName()),
+            Folder.getDefaultInstance(),
+            srcFolderName.getFolder());
+
+    FolderName dstFolderName = FolderName.of("_", bucket.getName(), UUID.randomUUID().toString());
+
+    RenameFolder.renameFolder(
+        bucket.getName(), srcFolderName.getFolder(), dstFolderName.getFolder());
+    try {
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(srcFolderName.toString());
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(dstFolderName.toString());
+      assertThrows(NotFoundException.class, () -> storageControl.getFolder(srcFolderName));
+    } finally {
+      storageControl.deleteFolder(dstFolderName);
+    }
+  }
+
+  @Test
+  public void deleteFolder() throws IOException {
+    FolderName folderName = FolderName.of("_", bucket.getName(), UUID.randomUUID().toString());
+    Folder gen1 =
+        storageControl.createFolder(
+            BucketName.of("_", bucket.getName()),
+            Folder.getDefaultInstance(),
+            folderName.getFolder());
+
+    DeleteFolder.deleteFolder(bucket.getName(), folderName.getFolder());
+    assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(folderName.toString());
+    assertThrows(NotFoundException.class, () -> storageControl.getFolder(folderName));
+  }
+
+  @Test
+  public void listFolder() throws IOException {
+    FolderName folderName = FolderName.of("_", bucket.getName(), UUID.randomUUID().toString());
+    Folder gen1 =
+        storageControl.createFolder(
+            BucketName.of("_", bucket.getName()),
+            Folder.getDefaultInstance(),
+            folderName.getFolder());
+
+    ListFolders.listFolders(bucket.getName());
+    try {
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(folderName.toString());
+    } finally {
+      storageControl.deleteFolder(folderName);
+    }
+  }
+}
diff --git a/samples/snippets/src/test/java/com/example/storage/control/v2/HNSTest.java b/samples/snippets/src/test/java/com/example/storage/control/v2/HNSTest.java
new file mode 100644
index 00000000..5d2ec83c
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/control/v2/HNSTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.example.storage.control.v2;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.example.storage.TestBase;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import org.junit.Test;
+
+public final class HNSTest extends TestBase {
+
+  @Test
+  public void createHierarchicalNamespaceBucket() throws Exception {
+    String newBucketName = RemoteStorageHelper.generateBucketName();
+    String projectId = storage.getOptions().getProjectId();
+    CreateHierarchicalNamespaceBucket.createHierarchicalNamespaceBucket(projectId, newBucketName);
+    try {
+      Bucket remoteBucket = storage.get(newBucketName);
+      assertThat(remoteBucket).isNotNull();
+      assertThat(remoteBucket.getHierarchicalNamespace().getEnabled()).isTrue();
+      assertThat(stdOut.getCapturedOutputAsUtf8String()).contains(newBucketName);
+    } finally {
+      storage.delete(newBucketName);
+    }
+  }
+}
diff --git a/samples/snippets/src/test/java/com/example/storage/managedfolders/CreateManagedFolderTest.java b/samples/snippets/src/test/java/com/example/storage/managedfolders/CreateManagedFolderTest.java
new file mode 100644
index 00000000..7bb8b74d
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/managedfolders/CreateManagedFolderTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.example.storage.managedfolders;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.cloud.testing.junit4.StdOutCaptureRule;
+import com.google.storage.control.v2.DeleteManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolderName;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class CreateManagedFolderTest {
+
+  @Rule public StdOutCaptureRule stdOut = new StdOutCaptureRule();
+
+  protected String bucketName;
+  protected Storage storage;
+  protected Bucket bucket;
+  protected String managedFolderId;
+  protected StorageControlClient storageControl;
+
+  @Before
+  public void setUp() throws IOException {
+    bucketName = RemoteStorageHelper.generateBucketName();
+    storageControl = StorageControlClient.create();
+    storage = StorageOptions.getDefaultInstance().getService();
+    managedFolderId = "new-managed-folder-" + UUID.randomUUID();
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName)
+            .setIamConfiguration(
+                IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+            .build();
+    bucket = storage.create(bucketInfo);
+  }
+
+  @After
+  public void tearDown() {
+    storageControl.deleteManagedFolder(
+        DeleteManagedFolderRequest.newBuilder()
+            .setName(ManagedFolderName.format("_", bucketName, managedFolderId))
+            .build());
+    storage.delete(bucketName);
+    storageControl.shutdown();
+  }
+
+  @Test
+  public void testCreateManagedFolder() throws Exception {
+    CreateManagedFolder.managedFolderCreate(bucketName, managedFolderId);
+    String got = stdOut.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format(managedFolderId));
+  }
+}
diff --git a/samples/snippets/src/test/java/com/example/storage/managedfolders/DeleteManagedFolderTest.java b/samples/snippets/src/test/java/com/example/storage/managedfolders/DeleteManagedFolderTest.java
new file mode 100644
index 00000000..7225d95f
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/managedfolders/DeleteManagedFolderTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.example.storage.managedfolders;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.cloud.testing.junit4.StdOutCaptureRule;
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.CreateManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class DeleteManagedFolderTest {
+
+  @Rule public StdOutCaptureRule stdOut = new StdOutCaptureRule();
+
+  protected String bucketName;
+  protected Storage storage;
+  protected Bucket bucket;
+  protected String managedFolderId;
+  protected StorageControlClient storageControl;
+
+  @Before
+  public void setUp() throws IOException {
+    bucketName = RemoteStorageHelper.generateBucketName();
+    storageControl = StorageControlClient.create();
+    storage = StorageOptions.getDefaultInstance().getService();
+    managedFolderId = "new-managed-folder-" + UUID.randomUUID();
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName)
+            .setIamConfiguration(
+                IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+            .build();
+    bucket = storage.create(bucketInfo);
+    storageControl.createManagedFolder(
+        CreateManagedFolderRequest.newBuilder()
+            // Set project to "_" to signify global bucket
+            .setParent(BucketName.format("_", bucketName))
+            .setManagedFolder(ManagedFolder.newBuilder().build())
+            .setManagedFolderId(managedFolderId)
+            .build());
+  }
+
+  @After
+  public void tearDown() {
+    storage.delete(bucketName);
+    storageControl.shutdown();
+  }
+
+  @Test
+  public void testDeleteManagedFolder() throws Exception {
+    DeleteManagedFolder.managedFolderDelete(bucketName, managedFolderId);
+    String got = stdOut.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format(managedFolderId));
+  }
+}
diff --git a/samples/snippets/src/test/java/com/example/storage/managedfolders/GetManagedFolderTest.java b/samples/snippets/src/test/java/com/example/storage/managedfolders/GetManagedFolderTest.java
new file mode 100644
index 00000000..3b5128cf
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/managedfolders/GetManagedFolderTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.example.storage.managedfolders;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.cloud.testing.junit4.StdOutCaptureRule;
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.CreateManagedFolderRequest;
+import com.google.storage.control.v2.DeleteManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.ManagedFolderName;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class GetManagedFolderTest {
+
+  @Rule public StdOutCaptureRule stdOut = new StdOutCaptureRule();
+
+  protected String bucketName;
+  protected Storage storage;
+  protected Bucket bucket;
+  protected String managedFolderId;
+  protected StorageControlClient storageControl;
+
+  @Before
+  public void setUp() throws IOException {
+    bucketName = RemoteStorageHelper.generateBucketName();
+    storageControl = StorageControlClient.create();
+    storage = StorageOptions.getDefaultInstance().getService();
+    managedFolderId = "new-managed-folder-" + UUID.randomUUID();
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName)
+            .setIamConfiguration(
+                IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+            .build();
+    bucket = storage.create(bucketInfo);
+    storageControl.createManagedFolder(
+        CreateManagedFolderRequest.newBuilder()
+            // Set project to "_" to signify global bucket
+            .setParent(BucketName.format("_", bucketName))
+            .setManagedFolder(ManagedFolder.newBuilder().build())
+            .setManagedFolderId(managedFolderId)
+            .build());
+  }
+
+  @After
+  public void tearDown() {
+    storageControl.deleteManagedFolder(
+        DeleteManagedFolderRequest.newBuilder()
+            .setName(ManagedFolderName.format("_", bucketName, managedFolderId))
+            .build());
+    storage.delete(bucketName);
+    storageControl.shutdown();
+  }
+
+  @Test
+  public void testGetManagedFolder() throws Exception {
+    GetManagedFolder.managedFolderGet(bucketName, managedFolderId);
+    String got = stdOut.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format(managedFolderId));
+  }
+}
diff --git a/samples/snippets/src/test/java/com/example/storage/managedfolders/ListManagedFolderTest.java b/samples/snippets/src/test/java/com/example/storage/managedfolders/ListManagedFolderTest.java
new file mode 100644
index 00000000..55bf404b
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/managedfolders/ListManagedFolderTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.example.storage.managedfolders;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.BucketInfo.IamConfiguration;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.cloud.testing.junit4.StdOutCaptureRule;
+import com.google.storage.control.v2.BucketName;
+import com.google.storage.control.v2.CreateManagedFolderRequest;
+import com.google.storage.control.v2.DeleteManagedFolderRequest;
+import com.google.storage.control.v2.ManagedFolder;
+import com.google.storage.control.v2.ManagedFolderName;
+import com.google.storage.control.v2.StorageControlClient;
+import java.io.IOException;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class ListManagedFolderTest {
+
+  @Rule public StdOutCaptureRule stdOut = new StdOutCaptureRule();
+
+  protected String bucketName;
+  protected Storage storage;
+  protected Bucket bucket;
+  protected String managedFolderId;
+  protected StorageControlClient storageControl;
+
+  @Before
+  public void setUp() throws IOException {
+    bucketName = RemoteStorageHelper.generateBucketName();
+    storageControl = StorageControlClient.create();
+    storage = StorageOptions.getDefaultInstance().getService();
+    managedFolderId = "new-managed-folder-" + UUID.randomUUID();
+    BucketInfo bucketInfo =
+        BucketInfo.newBuilder(bucketName)
+            .setIamConfiguration(
+                IamConfiguration.newBuilder().setIsUniformBucketLevelAccessEnabled(true).build())
+            .build();
+    bucket = storage.create(bucketInfo);
+    storageControl.createManagedFolder(
+        CreateManagedFolderRequest.newBuilder()
+            // Set project to "_" to signify global bucket
+            .setParent(BucketName.format("_", bucketName))
+            .setManagedFolder(ManagedFolder.newBuilder().build())
+            .setManagedFolderId(managedFolderId)
+            .build());
+  }
+
+  @After
+  public void tearDown() {
+    storageControl.deleteManagedFolder(
+        DeleteManagedFolderRequest.newBuilder()
+            .setName(ManagedFolderName.format("_", bucketName, managedFolderId))
+            .build());
+    storage.delete(bucketName);
+    storageControl.shutdown();
+  }
+
+  @Test
+  public void testListManagedFolder() throws Exception {
+    ListManagedFolders.managedFolderList(bucketName);
+    String got = stdOut.getCapturedOutputAsUtf8String();
+    assertThat(got).contains(String.format(managedFolderId));
+  }
+}