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

chore(deps): update dependency com.google.cloud:google-cloud-storage-control to v2.40.0
[![Mend Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com)

This PR contains the following updates:

| Package | Change | Age | Adoption | Passing | Confidence |
|---|---|---|---|---|---|
| [com.google.cloud:google-cloud-storage-control](https://togithub.com/googleapis/java-storage) | `2.38.0-beta` -> `2.40.0` | [![age](https://developer.mend.io/api/mc/badges/age/maven/com.google.cloud:google-cloud-storage-control/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://developer.mend.io/api/mc/badges/adoption/maven/com.google.cloud:google-cloud-storage-control/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://developer.mend.io/api/mc/badges/compatibility/maven/com.google.cloud:google-cloud-storage-control/2.38.0-beta/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://developer.mend.io/api/mc/badges/confidence/maven/com.google.cloud:google-cloud-storage-control/2.38.0-beta/2.40.0?slim=true)](https://docs.renovatebot.com/merge-confidence/) |

---

### Release Notes

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

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

##### Features

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

##### Bug Fixes

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

##### Dependencies

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

##### Documentation

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

</details>

---

### Configuration

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

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

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

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

---

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

---

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


dpebot: /gcbrun
dpebot: /gcbrun
forking-renovate[bot]: ### Edited/Blocked Notification

Renovate will not automatically rebase this PR, because it does not recognize the last commit author and assumes somebody else may have edited the PR.

You can manually request rebase by checking the rebase/retry box above.

 ⚠️ **Warning**: custom changes will be lost.
sydney-munro: /gcbrun
sydney-munro: /gcbrun

Patch:
diff --git a/README.md b/README.md
index 2b7d1dc6..f1ab2bff 100644
--- a/README.md
+++ b/README.md
@@ -46,7 +46,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.39.0</version>
+  <version>2.40.0</version>
 </dependency>
 <dependency>
   <groupId>com.google.cloud</groupId>
@@ -59,20 +59,20 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.40.0')
+implementation platform('com.google.cloud:libraries-bom:26.41.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.39.0'
+implementation 'com.google.cloud:google-cloud-storage:2.40.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.39.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.40.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -343,6 +343,12 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-storage/tree/
 | Set Public Access Prevention Enforced | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionEnforced.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionEnforced.java) |
 | Set Public Access Prevention Inherited | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionInherited.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetPublicAccessPreventionInherited.java) |
 | Set Retention Policy | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/bucket/SetRetentionPolicy.java) |
+| Create Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/CreateFolder.java) |
+| Create Hierarchical Namespace Bucket | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/CreateHierarchicalNamespaceBucket.java) |
+| Delete Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/DeleteFolder.java) |
+| Get Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/GetFolder.java) |
+| List Folders | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/ListFolders.java) |
+| Rename Folder | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/control/v2/RenameFolder.java) |
 | Activate Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/ActivateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/ActivateHmacKey.java) |
 | Create Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/CreateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/CreateHmacKey.java) |
 | Deactivate Hmac Key | [source code](https://github.com/googleapis/java-storage/blob/main/samples/snippets/src/main/java/com/example/storage/hmac/DeactivateHmacKey.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-storage&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/storage/hmac/DeactivateHmacKey.java) |
@@ -498,7 +504,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.39.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.40.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles
diff --git a/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/GapicBidiUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
index 19aba735..15278616 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
@@ -21,23 +21,28 @@ import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.BidiStreamingCallable;
+import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BidiWriteObjectResponse;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
+import io.grpc.Status;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 import java.util.function.Supplier;
@@ -47,18 +52,19 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
   private final BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write;
   private final RetryingDependencies deps;
   private final ResultRetryAlgorithm<?> alg;
-  private final String bucketName;
-  private final Supplier<GrpcCallContext> baseContextSupplier;
   private final SettableApiFuture<BidiWriteObjectResponse> resultFuture;
   private final ChunkSegmenter chunkSegmenter;
 
   private final BidiWriteCtx<BidiResumableWrite> writeCtx;
+  private final GrpcCallContext context;
   private final BidiObserver responseObserver;
 
   private volatile ApiStreamObserver<BidiWriteObjectRequest> stream;
   private boolean open = true;
   private boolean first = true;
   private boolean finished = false;
+  private volatile BidiWriteObjectRequest lastWrittenRequest;
+  private volatile RewindableContent currentContent;
 
   GapicBidiUnbufferedWritableByteChannel(
       BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write,
@@ -66,18 +72,18 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       ResultRetryAlgorithm<?> alg,
       SettableApiFuture<BidiWriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
-      BidiResumableWrite requestFactory,
+      BidiWriteCtx<BidiResumableWrite> writeCtx,
       Supplier<GrpcCallContext> baseContextSupplier) {
     this.write = write;
     this.deps = deps;
     this.alg = alg;
-    this.baseContextSupplier = baseContextSupplier;
-    this.bucketName = requestFactory.bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
-    this.writeCtx = new BidiWriteCtx<>(requestFactory);
+    this.writeCtx = writeCtx;
     this.responseObserver = new BidiObserver();
+    String bucketName = writeCtx.getRequestFactory().bucketName();
+    this.context = contextWithBucketName(bucketName, baseContextSupplier.get());
   }
 
   @Override
@@ -102,22 +108,22 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     if (!open) {
       return;
     }
-    ApiStreamObserver<BidiWriteObjectRequest> openedStream = openedStream();
-    if (!finished) {
-      BidiWriteObjectRequest message = finishMessage();
-      try {
-        openedStream.onNext(message);
-        finished = true;
-        openedStream.onCompleted();
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
+    try {
+      if (!finished) {
+        BidiWriteObjectRequest message = finishMessage();
+        lastWrittenRequest = message;
+        flush(Collections.singletonList(message));
+      } else {
+        if (stream != null) {
+          stream.onCompleted();
+          responseObserver.await();
+        }
       }
-    } else {
-      openedStream.onCompleted();
+    } finally {
+      open = false;
+      stream = null;
+      lastWrittenRequest = null;
     }
-    responseObserver.await();
-    open = false;
   }
 
   @VisibleForTesting
@@ -131,12 +137,18 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       throw new ClosedChannelException();
     }
 
-    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+    long begin = writeCtx.getConfirmedBytes().get();
+    currentContent = RewindableContent.of(srcs, srcsOffset, srcsLength);
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength, finalize);
+    if (data.length == 0) {
+      currentContent = null;
+      return 0;
+    }
 
     List<BidiWriteObjectRequest> messages = new ArrayList<>();
 
-    int bytesConsumed = 0;
-    for (ChunkSegment datum : data) {
+    for (int i = 0; i < data.length; i++) {
+      ChunkSegment datum = data[i];
       Crc32cLengthKnown crc32c = datum.getCrc32c();
       ByteString b = datum.getB();
       int contentSize = b.size();
@@ -149,11 +161,14 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       if (crc32c != null) {
         checksummedData.setCrc32C(crc32c.getValue());
       }
-      BidiWriteObjectRequest.Builder builder =
-          writeCtx
-              .newRequestBuilder()
-              .setWriteOffset(offset)
-              .setChecksummedData(checksummedData.build());
+      BidiWriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+      if (!first) {
+        builder.clearUploadId();
+        builder.clearObjectChecksums();
+      } else {
+        first = false;
+      }
+      builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
       if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
         if (cumulative != null) {
@@ -163,10 +178,17 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
         finished = true;
       }
 
-      BidiWriteObjectRequest build = possiblyPairDownBidiRequest(builder, first).build();
-      first = false;
+      if (i == data.length - 1 && !finished) {
+        if (finalize) {
+          builder.setFinishWrite(true);
+          finished = true;
+        } else {
+          builder.setFlush(true).setStateLookup(true);
+        }
+      }
+
+      BidiWriteObjectRequest build = builder.build();
       messages.add(build);
-      bytesConsumed += contentSize;
     }
     if (finalize && !finished) {
       messages.add(finishMessage());
@@ -176,10 +198,14 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     try {
       flush(messages);
     } catch (RuntimeException e) {
+      open = false;
       resultFuture.setException(e);
       throw e;
     }
 
+    long end = writeCtx.getConfirmedBytes().get();
+
+    long bytesConsumed = end - begin;
     return bytesConsumed;
   }
 
@@ -188,8 +214,11 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     long offset = writeCtx.getTotalSentBytes().get();
     Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
 
-    BidiWriteObjectRequest.Builder b =
-        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    BidiWriteObjectRequest.Builder b = writeCtx.newRequestBuilder();
+    if (!first) {
+      b.clearUploadId().clearObjectChecksums();
+    }
+    b.setFinishWrite(true).setWriteOffset(offset);
     if (crc32cValue != null) {
       b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
     }
@@ -201,13 +230,9 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
     if (stream == null) {
       synchronized (this) {
         if (stream == null) {
-          GrpcCallContext internalContext =
-              contextWithBucketName(bucketName, baseContextSupplier.get());
+          responseObserver.reset();
           stream =
-              this.write
-                  .withDefaultCallContext(internalContext)
-                  .bidiStreamingCall(responseObserver);
-          responseObserver.sem.drainPermits();
+              new GracefulOutboundStream(this.write.bidiStreamingCall(responseObserver, context));
         }
       }
     }
@@ -223,47 +248,28 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
             ApiStreamObserver<BidiWriteObjectRequest> opened = openedStream();
             for (BidiWriteObjectRequest message : segments) {
               opened.onNext(message);
+              lastWrittenRequest = message;
             }
-            if (!finished) {
-              BidiWriteObjectRequest message =
-                  BidiWriteObjectRequest.newBuilder().setFlush(true).setStateLookup(true).build();
-              opened.onNext(message);
+            if (lastWrittenRequest.getFinishWrite()) {
+              opened.onCompleted();
             }
             responseObserver.await();
             return null;
-          } catch (Exception e) {
+          } catch (Throwable t) {
             stream = null;
             first = true;
-            throw e;
+            t.addSuppressed(new AsyncStorageTaskException());
+            throw t;
           }
         },
         Decoder.identity());
   }
 
-  private static BidiWriteObjectRequest.Builder possiblyPairDownBidiRequest(
-      BidiWriteObjectRequest.Builder b, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && b.getWriteOffset() == 0) {
-      return b;
-    }
-
-    if (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
   private class BidiObserver implements ApiStreamObserver<BidiWriteObjectResponse> {
 
     private final Semaphore sem;
     private volatile BidiWriteObjectResponse last;
+    private volatile StorageException clientDetectedError;
     private volatile RuntimeException previousError;
 
     private BidiObserver() {
@@ -272,22 +278,96 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
 
     @Override
     public void onNext(BidiWriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        writeCtx.getConfirmedBytes().set((value.getPersistedSize()));
-      } else if (value.hasResource()) {
-        writeCtx.getConfirmedBytes().set(value.getResource().getSize());
+      boolean finalizing = lastWrittenRequest.getFinishWrite();
+      if (!finalizing && value.hasPersistedSize()) { // incremental
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long persistedSize = value.getPersistedSize();
+
+        if (totalSentBytes == persistedSize) {
+          writeCtx.getConfirmedBytes().set(persistedSize);
+          ok(value);
+        } else if (persistedSize < totalSentBytes) {
+          long delta = totalSentBytes - persistedSize;
+          // rewind our content and any state that my have run ahead of the actual ack'd bytes
+          currentContent.rewindTo(delta);
+          writeCtx.getTotalSentBytes().set(persistedSize);
+          writeCtx.getConfirmedBytes().set(persistedSize);
+          ok(value);
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_7.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else if (finalizing && value.hasResource()) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long finalSize = value.getResource().getSize();
+        if (totalSentBytes == finalSize) {
+          writeCtx.getConfirmedBytes().set(finalSize);
+          ok(value);
+        } else if (finalSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else if (!finalizing && value.hasResource()) {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_1.toStorageException(
+                ImmutableList.of(lastWrittenRequest), value, context, null));
+      } else if (finalizing && value.hasPersistedSize()) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long persistedSize = value.getPersistedSize();
+        // if a flush: true, state_lookup: true message is in the stream along with a
+        // finish_write: true, GCS can respond with the incremental update, gracefully handle this
+        // message
+        if (totalSentBytes == persistedSize) {
+          writeCtx.getConfirmedBytes().set(persistedSize);
+        } else if (persistedSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_3.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), value, context, null));
+        }
+      } else {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+                ImmutableList.of(lastWrittenRequest), value, context, null));
       }
-      sem.release();
-      last = value;
     }
 
     @Override
     public void onError(Throwable t) {
-      if (t instanceof RuntimeException) {
+      if (t instanceof OutOfRangeException) {
+        OutOfRangeException oore = (OutOfRangeException) t;
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                ImmutableList.of(lastWrittenRequest), null, context, oore));
+      } else if (t instanceof ApiException) {
+        // use StorageExceptions logic to translate from ApiException to our status codes ensuring
+        // things fall in line with our retry handlers.
+        // This is suboptimal, as it will initialize a second exception, however this is the
+        // unusual case, and it should not cause a significant overhead given its rarity.
+        StorageException tmp = StorageException.asStorageException((ApiException) t);
+        previousError =
+            ResumableSessionFailureScenario.toStorageException(
+                tmp.getCode(),
+                tmp.getMessage(),
+                tmp.getReason(),
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                t);
+        sem.release();
+      } else if (t instanceof RuntimeException) {
         previousError = (RuntimeException) t;
+        sem.release();
       }
-      sem.release();
     }
 
     @Override
@@ -298,6 +378,25 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       sem.release();
     }
 
+    private void ok(BidiWriteObjectResponse value) {
+      last = value;
+      sem.release();
+    }
+
+    private void clientDetectedError(StorageException storageException) {
+      open = false;
+      clientDetectedError = storageException;
+      // yes, check that previousError is not the same instance as e
+      if (previousError != null && previousError != storageException) {
+        storageException.addSuppressed(previousError);
+        previousError = null;
+      }
+      if (previousError == null) {
+        previousError = storageException;
+      }
+      sem.release();
+    }
+
     void await() {
       try {
         sem.acquire();
@@ -308,11 +407,69 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
           throw new RuntimeException(e);
         }
       }
+      StorageException e = clientDetectedError;
       RuntimeException err = previousError;
+      clientDetectedError = null;
+      previousError = null;
+      if ((e != null || err != null) && stream != null) {
+        if (lastWrittenRequest.getFinishWrite()) {
+          stream.onCompleted();
+        } else {
+          stream.onError(Status.CANCELLED.asRuntimeException());
+        }
+      }
+      if (e != null) {
+        throw e;
+      }
       if (err != null) {
-        previousError = null;
         throw err;
       }
     }
+
+    public void reset() {
+      sem.drainPermits();
+      last = null;
+      clientDetectedError = null;
+      previousError = null;
+    }
+  }
+
+  /**
+   * Prevent "already half-closed" if we previously called onComplete but then detect an error and
+   * call onError
+   */
+  private static final class GracefulOutboundStream
+      implements ApiStreamObserver<BidiWriteObjectRequest> {
+
+    private final ApiStreamObserver<BidiWriteObjectRequest> delegate;
+    private volatile boolean closing;
+
+    private GracefulOutboundStream(ApiStreamObserver<BidiWriteObjectRequest> delegate) {
+      this.delegate = delegate;
+      this.closing = false;
+    }
+
+    @Override
+    public void onNext(BidiWriteObjectRequest value) {
+      delegate.onNext(value);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      if (closing) {
+        return;
+      }
+      closing = true;
+      delegate.onError(t);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (closing) {
+        return;
+      }
+      closing = true;
+      delegate.onCompleted();
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
index 536eba7f..05387326 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
@@ -155,7 +155,7 @@ final class GapicBidiWritableByteChannelSessionBuilder {
                             resultFuture,
                             new ChunkSegmenter(
                                 boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-                            start,
+                            new BidiWriteCtx<>(start),
                             Retrying::newCallContext))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
index 16c6ec0a..984c7bfd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
@@ -20,11 +20,13 @@ import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
 
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -33,11 +35,7 @@ import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.ExecutionException;
-import java.util.function.Consumer;
-import java.util.function.LongConsumer;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
@@ -55,13 +53,14 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
   private boolean open = true;
   private boolean first = true;
   private boolean finished = false;
+  private volatile WriteObjectRequest lastWrittenRequest;
 
   GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      ResumableWrite requestFactory) {
-    String bucketName = requestFactory.bucketName();
+      WriteCtx<ResumableWrite> writeCtx) {
+    String bucketName = writeCtx.getRequestFactory().bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
@@ -69,8 +68,8 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
         contextWithBucketName(bucketName, GrpcCallContext.createDefault());
     this.write = write.withDefaultCallContext(internalContext);
 
-    this.writeCtx = new WriteCtx<>(requestFactory);
-    this.responseObserver = new Observer(writeCtx.getConfirmedBytes()::set, resultFuture::set);
+    this.writeCtx = writeCtx;
+    this.responseObserver = new Observer(internalContext);
   }
 
   @Override
@@ -92,27 +91,24 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
 
   @Override
   public void close() throws IOException {
+    if (!open) {
+      return;
+    }
+    open = false;
     ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
-    if (!finished) {
-      WriteObjectRequest message = finishMessage();
-      try {
+    try {
+      if (!finished) {
+        WriteObjectRequest message = finishMessage();
+        lastWrittenRequest = message;
         openedStream.onNext(message);
-        openedStream.onCompleted();
         finished = true;
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
-      }
-    } else {
-      try {
-        openedStream.onCompleted();
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
       }
+      openedStream.onCompleted();
+      responseObserver.await();
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
     }
-    open = false;
-    responseObserver.await();
   }
 
   private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
@@ -122,51 +118,54 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
     }
 
     ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
-
-    List<WriteObjectRequest> messages = new ArrayList<>();
+    if (data.length == 0) {
+      return 0;
+    }
 
     ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
     int bytesConsumed = 0;
-    for (ChunkSegment datum : data) {
-      Crc32cLengthKnown crc32c = datum.getCrc32c();
-      ByteString b = datum.getB();
-      int contentSize = b.size();
-      long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
-      Crc32cLengthKnown cumulative =
-          writeCtx
-              .getCumulativeCrc32c()
-              .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
-      ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
-      if (crc32c != null) {
-        checksummedData.setCrc32C(crc32c.getValue());
-      }
-      WriteObjectRequest.Builder builder =
-          writeCtx
-              .newRequestBuilder()
-              .setWriteOffset(offset)
-              .setChecksummedData(checksummedData.build());
-      if (!datum.isOnlyFullBlocks()) {
-        builder.setFinishWrite(true);
-        if (cumulative != null) {
-          builder.setObjectChecksums(
-              ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+    try {
+      for (int i = 0; i < data.length; i++) {
+        ChunkSegment datum = data[i];
+        Crc32cLengthKnown crc32c = datum.getCrc32c();
+        ByteString b = datum.getB();
+        int contentSize = b.size();
+        long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+        Crc32cLengthKnown cumulative =
+            writeCtx
+                .getCumulativeCrc32c()
+                .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
+        ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
+        if (crc32c != null) {
+          checksummedData.setCrc32C(crc32c.getValue());
+        }
+        WriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+        if (!first) {
+          builder.clearUploadId();
+          builder.clearWriteObjectSpec();
+          builder.clearObjectChecksums();
+        }
+        builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
+        if (!datum.isOnlyFullBlocks() || (finalize && i + 1 == data.length)) {
+          builder.setFinishWrite(true);
+          if (cumulative != null) {
+            builder.setObjectChecksums(
+                ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+          }
+          finished = true;
         }
-        finished = true;
-      }
-
-      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
-      first = false;
-      messages.add(build);
-      bytesConsumed += contentSize;
-    }
-    if (finalize && !finished) {
-      messages.add(finishMessage());
-      finished = true;
-    }
 
-    try {
-      for (WriteObjectRequest message : messages) {
-        openedStream.onNext(message);
+        WriteObjectRequest build = builder.build();
+        first = false;
+        lastWrittenRequest = build;
+        openedStream.onNext(build);
+        bytesConsumed += contentSize;
+      }
+      if (finalize && !finished) {
+        WriteObjectRequest finishMessage = finishMessage();
+        lastWrittenRequest = finishMessage;
+        openedStream.onNext(finishMessage);
+        finished = true;
       }
     } catch (RuntimeException e) {
       resultFuture.setException(e);
@@ -201,73 +200,104 @@ final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
     return stream;
   }
 
-  /**
-   * Several fields of a WriteObjectRequest are only allowed on the "first" message sent to gcs,
-   * this utility method centralizes the logic necessary to clear those fields for use by subsequent
-   * messages.
-   */
-  private static WriteObjectRequest.Builder possiblyPairDownRequest(
-      WriteObjectRequest.Builder b, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && b.getWriteOffset() == 0) {
-      return b;
-    }
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
-  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+  class Observer implements ApiStreamObserver<WriteObjectResponse> {
 
-    private final LongConsumer sizeCallback;
-    private final Consumer<WriteObjectResponse> completeCallback;
+    private final GrpcCallContext context;
 
     private final SettableApiFuture<Void> invocationHandle;
     private volatile WriteObjectResponse last;
 
-    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
+    Observer(GrpcCallContext context) {
+      this.context = context;
       this.invocationHandle = SettableApiFuture.create();
     }
 
     @Override
     public void onNext(WriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        sizeCallback.accept(value.getPersistedSize());
-      } else if (value.hasResource()) {
-        sizeCallback.accept(value.getResource().getSize());
-      }
       last = value;
     }
 
-    /**
-     * observed exceptions so far
-     *
-     * <ol>
-     *   <li>{@link com.google.api.gax.rpc.OutOfRangeException}
-     *   <li>{@link com.google.api.gax.rpc.AlreadyExistsException}
-     *   <li>{@link io.grpc.StatusRuntimeException}
-     * </ol>
-     */
     @Override
     public void onError(Throwable t) {
-      invocationHandle.setException(t);
+      if (t instanceof ApiException) {
+        // use StorageExceptions logic to translate from ApiException to our status codes ensuring
+        // things fall in line with our retry handlers.
+        // This is suboptimal, as it will initialize a second exception, however this is the
+        // unusual case, and it should not cause a significant overhead given its rarity.
+        StorageException tmp = StorageException.asStorageException((ApiException) t);
+        StorageException storageException =
+            ResumableSessionFailureScenario.toStorageException(
+                tmp.getCode(),
+                tmp.getMessage(),
+                tmp.getReason(),
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                t);
+        resultFuture.setException(storageException);
+        invocationHandle.setException(storageException);
+      } else {
+        resultFuture.setException(t);
+        invocationHandle.setException(t);
+      }
     }
 
     @Override
     public void onCompleted() {
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
+      boolean finalizing = lastWrittenRequest.getFinishWrite();
+      if (last == null) {
+        clientDetectedError(
+            ResumableSessionFailureScenario.toStorageException(
+                0,
+                "onComplete without preceding onNext, unable to determine success.",
+                "invalid",
+                ImmutableList.of(lastWrittenRequest),
+                null,
+                context,
+                null));
+      } else if (last.hasResource() /* && finalizing*/) {
+        long totalSentBytes = writeCtx.getTotalSentBytes().get();
+        long finalSize = last.getResource().getSize();
+        if (totalSentBytes == finalSize) {
+          ok(finalSize);
+        } else if (finalSize < totalSentBytes) {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), last, context, null));
+        } else {
+          clientDetectedError(
+              ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+                  ImmutableList.of(lastWrittenRequest), last, context, null));
+        }
+      } else if (!finalizing || last.hasPersistedSize()) { // unexpected incremental response
+        clientDetectedError(
+            ResumableSessionFailureScenario.toStorageException(
+                0,
+                "Unexpected incremental response for finalizing request.",
+                "invalid",
+                ImmutableList.of(lastWrittenRequest),
+                last,
+                context,
+                null));
+      } else {
+        clientDetectedError(
+            ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+                ImmutableList.of(lastWrittenRequest), last, context, null));
       }
+    }
+
+    private void ok(long persistedSize) {
+      writeCtx.getConfirmedBytes().set(persistedSize);
+      resultFuture.set(last);
       invocationHandle.set(null);
     }
 
+    private void clientDetectedError(StorageException storageException) {
+      open = false;
+      resultFuture.setException(storageException);
+      invocationHandle.setException(storageException);
+    }
+
     void await() {
       try {
         invocationHandle.get();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index f4a8afdb..1e02f968 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -305,7 +305,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, start);
+                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
                 })
                 .andThen(StorageByteChannels.writable()::createSynchronized));
@@ -346,7 +346,7 @@ final class GapicWritableByteChannelSessionBuilder {
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, start);
+                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
                 })
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 38acf828..92081e03 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -600,7 +600,13 @@ public final class GrpcStorageOptions extends StorageOptions
     @BetaApi
     @Override
     public GrpcStorageOptions build() {
-      return new GrpcStorageOptions(this, defaults());
+      GrpcStorageOptions options = new GrpcStorageOptions(this, defaults());
+      // TODO: Remove when https://github.com/googleapis/sdk-platform-java/issues/2911 is resolved
+      if (options.getUniverseDomain() != null) {
+        this.setHost("https://storage." + options.getUniverseDomain());
+        return new GrpcStorageOptions(this, defaults());
+      }
+      return options;
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
index 8c06d70d..88dbaf7b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
@@ -26,11 +26,11 @@ import com.google.api.gax.rpc.ApiException;
 import com.google.cloud.BaseServiceException;
 import com.google.cloud.storage.StorageException.IOExceptionCallable;
 import com.google.common.io.CharStreams;
-import com.google.protobuf.MessageOrBuilder;
+import com.google.protobuf.Message;
+import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
 import com.google.storage.v2.WriteObjectRequest;
-import com.google.storage.v2.WriteObjectResponse;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -81,10 +81,6 @@ enum ResumableSessionFailureScenario {
   private static final String PREFIX_I = "\t|< ";
   private static final String PREFIX_O = "\t|> ";
   private static final String PREFIX_X = "\t|  ";
-  // define some constants for tab widths that are more compressed that the literals
-  private static final String T1 = "\t";
-  private static final String T2 = "\t\t";
-  private static final String T3 = "\t\t\t";
 
   private static final Predicate<String> includedHeaders =
       matches("Content-Length")
@@ -134,8 +130,10 @@ enum ResumableSessionFailureScenario {
   }
 
   StorageException toStorageException(
-      @NonNull List<@NonNull WriteObjectRequest> reqs,
-      @Nullable WriteObjectResponse resp,
+      /*In java List<WriteObjectRequest> is not a sub-type of List<Message> despite WriteObjectRequest being a Message.
+       * intentionally only define List so the compiler doesn't complain */
+      @SuppressWarnings("rawtypes") @NonNull List reqs,
+      @Nullable Message resp,
       @NonNull GrpcCallContext context,
       @Nullable Throwable cause) {
     return toStorageException(code, message, reason, reqs, resp, context, cause);
@@ -161,8 +159,8 @@ enum ResumableSessionFailureScenario {
       int code,
       String message,
       @Nullable String reason,
-      @NonNull List<@NonNull WriteObjectRequest> reqs,
-      @Nullable WriteObjectResponse resp,
+      @NonNull List reqs,
+      @Nullable Message resp,
       @NonNull GrpcCallContext context,
       @Nullable Throwable cause) {
     final StringBuilder sb = new StringBuilder();
@@ -177,35 +175,8 @@ enum ResumableSessionFailureScenario {
       } else {
         sb.append(",");
       }
-      WriteObjectRequest req = reqs.get(i);
-      sb.append("\n").append(PREFIX_O).append(T1).append(req.getClass().getName()).append("{");
-      if (req.hasUploadId()) {
-        sb.append("\n").append(PREFIX_O).append(T2).append("upload_id: ").append(req.getUploadId());
-      }
-      long writeOffset = req.getWriteOffset();
-      if (req.hasChecksummedData()) {
-        ChecksummedData checksummedData = req.getChecksummedData();
-        sb.append("\n").append(PREFIX_O).append(T2);
-        sb.append(
-            String.format(
-                "checksummed_data: {range: [%d:%d]",
-                writeOffset, writeOffset + checksummedData.getContent().size()));
-        if (checksummedData.hasCrc32C()) {
-          sb.append(", crc32c: ").append(checksummedData.getCrc32C());
-        }
-        sb.append("}");
-      } else {
-        sb.append("\n").append(PREFIX_O).append(T2).append("write_offset: ").append(writeOffset);
-      }
-      if (req.getFinishWrite()) {
-        sb.append("\n").append(PREFIX_O).append(T2).append("finish_write: true");
-      }
-      if (req.hasObjectChecksums()) {
-        ObjectChecksums objectChecksums = req.getObjectChecksums();
-        sb.append("\n").append(PREFIX_O).append(T2).append("object_checksums: ").append("{");
-        fmt(objectChecksums, PREFIX_O, T3, sb);
-        sb.append("\n").append(PREFIX_O).append(T2).append("}");
-      }
+      Message req = (Message) reqs.get(i);
+      fmt(req, PREFIX_O, Indentation.T1, sb);
       sb.append("\n").append(PREFIX_O).append("\t}");
       if (i == length - 1) {
         sb.append("\n").append(PREFIX_O).append("]");
@@ -217,7 +188,7 @@ enum ResumableSessionFailureScenario {
     // response context
     if (resp != null) {
       sb.append("\n").append(PREFIX_I).append(resp.getClass().getName()).append("{");
-      fmt(resp, PREFIX_I, T1, sb);
+      fmt(resp, PREFIX_I, Indentation.T1, sb);
       sb.append("\n").append(PREFIX_I).append("}");
       sb.append("\n").append(PREFIX_X);
     }
@@ -250,7 +221,8 @@ enum ResumableSessionFailureScenario {
         sb.append("\n").append(PREFIX_X);
       }
     }
-    return new StorageException(code, sb.toString(), reason, cause);
+    StorageException se = new StorageException(code, sb.toString(), reason, cause);
+    return se;
   }
 
   static StorageException toStorageException(
@@ -359,16 +331,122 @@ enum ResumableSessionFailureScenario {
   }
 
   private static void fmt(
-      MessageOrBuilder msg,
+      Message msg,
       @SuppressWarnings("SameParameterValue") String prefix,
-      String indentation,
+      Indentation indentation,
       StringBuilder sb) {
-    String string = msg.toString();
-    // drop the final new line before prefixing
-    string = string.replaceAll("\n$", "");
-    sb.append("\n")
-        .append(prefix)
-        .append(indentation)
-        .append(string.replaceAll("\r?\n", "\n" + prefix + indentation));
+    if (msg instanceof WriteObjectRequest) {
+      WriteObjectRequest req = (WriteObjectRequest) msg;
+      fmtWriteObjectRequest(req, prefix, indentation, sb);
+    } else if (msg instanceof BidiWriteObjectRequest) {
+      BidiWriteObjectRequest req = (BidiWriteObjectRequest) msg;
+      fmtBidiWriteObjectRequest(req, prefix, indentation, sb);
+    } else {
+      String string = msg.toString();
+      // drop the final new line before prefixing
+      string = string.replaceAll("\n$", "");
+      sb.append("\n")
+          .append(prefix)
+          .append(indentation)
+          .append(string.replaceAll("\r?\n", "\n" + prefix + indentation.indentation));
+    }
+  }
+
+  private static void fmtWriteObjectRequest(
+      WriteObjectRequest req, String prefix, Indentation t1, StringBuilder sb) {
+    Indentation t2 = t1.indent();
+    Indentation t3 = t2.indent();
+    sb.append("\n").append(prefix).append(t1).append(req.getClass().getName()).append("{");
+    if (req.hasUploadId()) {
+      sb.append("\n").append(prefix).append(t2).append("upload_id: ").append(req.getUploadId());
+    }
+    long writeOffset = req.getWriteOffset();
+    if (req.hasChecksummedData()) {
+      ChecksummedData checksummedData = req.getChecksummedData();
+      sb.append("\n").append(prefix).append(t2);
+      sb.append(
+          String.format(
+              "checksummed_data: {range: [%d:%d]",
+              writeOffset, writeOffset + checksummedData.getContent().size()));
+      if (checksummedData.hasCrc32C()) {
+        sb.append(", crc32c: ").append(checksummedData.getCrc32C());
+      }
+      sb.append("}");
+    } else {
+      sb.append("\n").append(prefix).append(t2).append("write_offset: ").append(writeOffset);
+    }
+    if (req.getFinishWrite()) {
+      sb.append("\n").append(prefix).append(t2).append("finish_write: true");
+    }
+    if (req.hasObjectChecksums()) {
+      ObjectChecksums objectChecksums = req.getObjectChecksums();
+      sb.append("\n").append(prefix).append(t2).append("object_checksums: ").append("{");
+      fmt(objectChecksums, prefix, t3, sb);
+      sb.append("\n").append(prefix).append(t2).append("}");
+    }
+  }
+
+  private static void fmtBidiWriteObjectRequest(
+      BidiWriteObjectRequest req, String prefix, Indentation t1, StringBuilder sb) {
+    Indentation t2 = t1.indent();
+    Indentation t3 = t2.indent();
+    sb.append("\n").append(prefix).append(t1).append(req.getClass().getName()).append("{");
+    if (req.hasUploadId()) {
+      sb.append("\n").append(prefix).append(t2).append("upload_id: ").append(req.getUploadId());
+    }
+    long writeOffset = req.getWriteOffset();
+    if (req.hasChecksummedData()) {
+      ChecksummedData checksummedData = req.getChecksummedData();
+      sb.append("\n").append(prefix).append(t2);
+      sb.append(
+          String.format(
+              "checksummed_data: {range: [%d:%d]",
+              writeOffset, writeOffset + checksummedData.getContent().size()));
+      if (checksummedData.hasCrc32C()) {
+        sb.append(", crc32c: ").append(checksummedData.getCrc32C());
+      }
+      sb.append("}");
+    } else {
+      sb.append("\n").append(prefix).append(t2).append("write_offset: ").append(writeOffset);
+    }
+    if (req.getFlush()) {
+      sb.append("\n").append(prefix).append(t2).append("flush: true");
+    }
+    if (req.getStateLookup()) {
+      sb.append("\n").append(prefix).append(t2).append("state_lookup: true");
+    }
+    if (req.getFinishWrite()) {
+      sb.append("\n").append(prefix).append(t2).append("finish_write: true");
+    }
+    if (req.hasObjectChecksums()) {
+      ObjectChecksums objectChecksums = req.getObjectChecksums();
+      sb.append("\n").append(prefix).append(t2).append("object_checksums: ").append("{");
+      fmt(objectChecksums, prefix, t3, sb);
+      sb.append("\n").append(prefix).append(t2).append("}");
+    }
+  }
+
+  enum Indentation {
+    T1("\t"),
+    T2("\t\t"),
+    T3("\t\t\t"),
+    T4("\t\t\t\t"),
+    ;
+
+    private final String indentation;
+
+    Indentation(String indentation) {
+      this.indentation = indentation;
+    }
+
+    Indentation indent() {
+      int ordinal = ordinal();
+      return values()[ordinal + 1];
+    }
+
+    @Override
+    public String toString() {
+      return indentation;
+    }
   }
 }
diff --git a/pom.xml b/pom.xml
index 67017b33..280a4196 100644
--- a/pom.xml
+++ b/pom.xml
@@ -81,7 +81,7 @@
       <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>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 1388f8c4..4d9ad4ee 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,7 +30,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.39.0</version>
+      <version>2.40.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
@@ -72,7 +72,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.38.0-beta</version>
+      <version>2.40.0</version>
       <scope>compile</scope>
     </dependency>
   </dependencies>


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