Repository: googleapis/java-storage
Pull request #2527 (2024-05-06 22:30:16)

fix: add strict client side response validation for gRPC chunked resumable uploads
* Rename JsonResumableSessionFailureScenario to ResumableSessionFailureScenario. The failure scenarios themselves are not json specific, and the methods which are json specific can have grpc overloads
* Add more tests to validate GapicUnbufferedChunkedResumableWritableByteChannel is able to properly detect and handle various success responses from GCS which are not success for the client.





Patch:
diff --git a/google-cloud-storage-bom/pom.xml b/google-cloud-storage-bom/pom.xml
index d7e347cc..69282077 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -89,17 +89,17 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
     </dependencies>
   </dependencyManagement>
diff --git a/google-cloud-storage-control/pom.xml b/google-cloud-storage-control/pom.xml
index fb0929bc..314f91b9 100644
--- a/google-cloud-storage-control/pom.xml
+++ b/google-cloud-storage-control/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-control</artifactId>
-    <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+    <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
     <name>google-cloud-storage-control</name>
     <description>GRPC library for google-cloud-storage-control</description>
     <parent>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
index aa48810d..bbcd986a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
@@ -23,6 +23,7 @@ import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
+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;
@@ -41,10 +42,9 @@ 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 java.util.function.Supplier;
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class GapicUnbufferedChunkedResumableWritableByteChannel
     implements UnbufferedWritableByteChannel {
@@ -58,30 +58,26 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
   private final RetryingDependencies deps;
   private final ResultRetryAlgorithm<?> alg;
   private final Supplier<GrpcCallContext> baseContextSupplier;
-  private final LongConsumer sizeCallback;
-  private final Consumer<WriteObjectResponse> completeCallback;
 
-  private boolean open = true;
+  private volatile boolean open = true;
   private boolean finished = false;
 
   GapicUnbufferedChunkedResumableWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       @NonNull ChunkSegmenter chunkSegmenter,
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      ResumableWrite requestFactory,
+      WriteCtx<ResumableWrite> writeCtx,
       RetryingDependencies deps,
       ResultRetryAlgorithm<?> alg,
       Supplier<GrpcCallContext> baseContextSupplier) {
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
     this.write = write;
-    this.bucketName = requestFactory.bucketName();
-    this.writeCtx = new WriteCtx<>(requestFactory);
+    this.bucketName = writeCtx.getRequestFactory().bucketName();
+    this.writeCtx = writeCtx;
     this.deps = deps;
     this.alg = alg;
     this.baseContextSupplier = baseContextSupplier;
-    this.sizeCallback = writeCtx.getConfirmedBytes()::set;
-    this.completeCallback = resultFuture::set;
   }
 
   @Override
@@ -106,7 +102,7 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
     if (open && !finished) {
       WriteObjectRequest message = finishMessage(true);
       try {
-        flush(ImmutableList.of(message));
+        flush(ImmutableList.of(message), null, true);
         finished = true;
       } catch (RuntimeException e) {
         resultFuture.setException(e);
@@ -122,12 +118,13 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
       throw new ClosedChannelException();
     }
 
+    long begin = writeCtx.getConfirmedBytes().get();
+    RewindableContent content = RewindableContent.of(srcs, srcsOffset, srcsLength);
     ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
 
     List<WriteObjectRequest> messages = new ArrayList<>();
 
     boolean first = true;
-    int bytesConsumed = 0;
     for (ChunkSegment datum : data) {
       Crc32cLengthKnown crc32c = datum.getCrc32c();
       ByteString b = datum.getB();
@@ -144,8 +141,13 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
       WriteObjectRequest.Builder builder =
           writeCtx
               .newRequestBuilder()
+              .clearWriteObjectSpec()
+              .clearObjectChecksums()
               .setWriteOffset(offset)
               .setChecksummedData(checksummedData.build());
+      if (!first) {
+        builder.clearUploadId();
+      }
       if (!datum.isOnlyFullBlocks()) {
         builder.setFinishWrite(true);
         if (cumulative != null) {
@@ -155,10 +157,9 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
         finished = true;
       }
 
-      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
+      WriteObjectRequest build = builder.build();
       first = false;
       messages.add(build);
-      bytesConsumed += contentSize;
     }
     if (finalize && !finished) {
       messages.add(finishMessage(first));
@@ -166,12 +167,15 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
     }
 
     try {
-      flush(messages);
+      flush(messages, content, finalize);
     } catch (RuntimeException e) {
       resultFuture.setException(e);
       throw e;
     }
 
+    long end = writeCtx.getConfirmedBytes().get();
+
+    long bytesConsumed = end - begin;
     return bytesConsumed;
   }
 
@@ -182,14 +186,20 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
 
     WriteObjectRequest.Builder b =
         writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    if (!first) {
+      b.clearUploadId();
+    }
     if (crc32cValue != null) {
       b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
     }
-    WriteObjectRequest message = possiblyPairDownRequest(b, first).build();
+    WriteObjectRequest message = b.build();
     return message;
   }
 
-  private void flush(@NonNull List<WriteObjectRequest> segments) {
+  private void flush(
+      @NonNull List<WriteObjectRequest> segments,
+      @Nullable RewindableContent content,
+      boolean finalizing) {
     GrpcCallContext internalContext = contextWithBucketName(bucketName, baseContextSupplier.get());
     ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> callable =
         write.withDefaultCallContext(internalContext);
@@ -198,7 +208,7 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
         deps,
         alg,
         () -> {
-          Observer observer = new Observer(sizeCallback, completeCallback);
+          Observer observer = new Observer(content, finalizing);
           ApiStreamObserver<WriteObjectRequest> write = callable.clientStreamingCall(observer);
 
           for (WriteObjectRequest message : segments) {
@@ -211,81 +221,93 @@ final class GapicUnbufferedChunkedResumableWritableByteChannel
         Decoder.identity());
   }
 
-  /**
-   * 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 (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (b.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b;
-  }
-
   @VisibleForTesting
   WriteCtx<?> getWriteCtx() {
     return writeCtx;
   }
 
-  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+  class Observer implements ApiStreamObserver<WriteObjectResponse> {
 
-    private final LongConsumer sizeCallback;
-    private final Consumer<WriteObjectResponse> completeCallback;
+    private final RewindableContent content;
+    private final boolean finalizing;
 
     private final SettableApiFuture<Void> invocationHandle;
     private volatile WriteObjectResponse last;
 
-    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
+    Observer(@Nullable RewindableContent content, boolean finalizing) {
+      this.content = content;
+      this.finalizing = finalizing;
       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 OutOfRangeException) {
+        OutOfRangeException oore = (OutOfRangeException) t;
+        open = false;
+        invocationHandle.setException(
+            ResumableSessionFailureScenario.SCENARIO_5.toStorageException());
+      } else {
+        invocationHandle.setException(t);
+      }
     }
 
     @Override
     public void onCompleted() {
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
+      try {
+        if (last == null) {
+          throw new StorageException(
+              0, "onComplete without preceding onNext, unable to determine success.");
+        } else if (!finalizing && last.hasPersistedSize()) { // incremental
+          long totalSentBytes = writeCtx.getTotalSentBytes().get();
+          long persistedSize = last.getPersistedSize();
+
+          if (totalSentBytes == persistedSize) {
+            writeCtx.getConfirmedBytes().set(persistedSize);
+          } 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
+            content.rewindTo(delta);
+            writeCtx.getTotalSentBytes().set(persistedSize);
+            writeCtx.getConfirmedBytes().set(persistedSize);
+          } else {
+            throw ResumableSessionFailureScenario.SCENARIO_7.toStorageException();
+          }
+        } else if (finalizing && last.hasResource()) {
+          long totalSentBytes = writeCtx.getTotalSentBytes().get();
+          long finalSize = last.getResource().getSize();
+          if (totalSentBytes == finalSize) {
+            writeCtx.getConfirmedBytes().set(finalSize);
+            resultFuture.set(last);
+          } else if (finalSize < totalSentBytes) {
+            throw ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException();
+          } else {
+            throw ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException();
+          }
+        } else if (!finalizing && last.hasResource()) {
+          throw ResumableSessionFailureScenario.SCENARIO_1.toStorageException();
+        } else if (finalizing && last.hasPersistedSize()) {
+          long totalSentBytes = writeCtx.getTotalSentBytes().get();
+          long persistedSize = last.getPersistedSize();
+          if (persistedSize < totalSentBytes) {
+            throw ResumableSessionFailureScenario.SCENARIO_3.toStorageException();
+          } else {
+            throw ResumableSessionFailureScenario.SCENARIO_2.toStorageException();
+          }
+        } else {
+          throw ResumableSessionFailureScenario.SCENARIO_0.toStorageException();
+        }
+      } catch (Throwable se) {
+        open = false;
+        invocationHandle.setException(se);
+      } finally {
+        invocationHandle.set(null);
       }
-      invocationHandle.set(null);
     }
 
     void await() {
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 1c6ad18c..88540533 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
@@ -299,13 +299,13 @@ final class GapicWritableByteChannelSessionBuilder {
                         result,
                         getChunkSegmenter(),
                         write,
-                        ResumableWrite.identity(start),
+                        new WriteCtx<>(start),
                         deps,
                         alg,
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, ResumableWrite.identity(start));
+                        result, getChunkSegmenter(), write, start);
                   }
                 })
                 .andThen(StorageByteChannels.writable()::createSynchronized));
@@ -340,13 +340,13 @@ final class GapicWritableByteChannelSessionBuilder {
                         result,
                         getChunkSegmenter(),
                         write,
-                        ResumableWrite.identity(start),
+                        new WriteCtx<>(start),
                         deps,
                         alg,
                         Retrying::newCallContext);
                   } else {
                     return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, ResumableWrite.identity(start));
+                        result, getChunkSegmenter(), write, start);
                   }
                 })
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
index 622988d4..9d3afa5a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
@@ -101,7 +101,7 @@ final class JsonResumableSessionPutTask
 
       int code = response.getStatusCode();
 
-      if (!finalizing && JsonResumableSessionFailureScenario.isContinue(code)) {
+      if (!finalizing && ResumableSessionFailureScenario.isContinue(code)) {
         long effectiveEnd = ((HttpContentRange.HasRange<?>) contentRange).range().endOffset();
         @Nullable String range = response.getHeaders().getRange();
         ByteRangeSpec ackRange = ByteRangeSpec.parse(range);
@@ -114,11 +114,11 @@ final class JsonResumableSessionPutTask
           return ResumableOperationResult.incremental(ackRange.endOffset());
         } else {
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);
+              ResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
         }
-      } else if (finalizing && JsonResumableSessionFailureScenario.isOk(code)) {
+      } else if (finalizing && ResumableSessionFailureScenario.isOk(code)) {
         @Nullable StorageObject storageObject;
         BigInteger actualSize = BigInteger.ZERO;
 
@@ -145,7 +145,7 @@ final class JsonResumableSessionPutTask
         } else {
           response.ignore();
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
+              ResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
                   uploadId, response, null, () -> null);
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
@@ -158,35 +158,35 @@ final class JsonResumableSessionPutTask
           return ResumableOperationResult.complete(storageObject, actualSize.longValue());
         } else if (compare > 0) {
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
+              ResumableSessionFailureScenario.SCENARIO_4_1.toStorageException(
                   uploadId, response, null, toString(storageObject));
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
         } else {
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
+              ResumableSessionFailureScenario.SCENARIO_4_2.toStorageException(
                   uploadId, response, null, toString(storageObject));
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
         }
-      } else if (!finalizing && JsonResumableSessionFailureScenario.isOk(code)) {
+      } else if (!finalizing && ResumableSessionFailureScenario.isOk(code)) {
         StorageException se =
-            JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException(uploadId, response);
+            ResumableSessionFailureScenario.SCENARIO_1.toStorageException(uploadId, response);
         span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
         throw se;
-      } else if (finalizing && JsonResumableSessionFailureScenario.isContinue(code)) {
+      } else if (finalizing && ResumableSessionFailureScenario.isContinue(code)) {
         // in order to finalize the content range must have a size, cast down to read it
         HttpContentRange.HasSize size = (HttpContentRange.HasSize) contentRange;
 
         ByteRangeSpec range = ByteRangeSpec.parse(response.getHeaders().getRange());
         if (range.endOffsetInclusive() < size.getSize()) {
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_3.toStorageException(uploadId, response);
+              ResumableSessionFailureScenario.SCENARIO_3.toStorageException(uploadId, response);
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
         } else {
           StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_2.toStorageException(uploadId, response);
+              ResumableSessionFailureScenario.SCENARIO_2.toStorageException(uploadId, response);
           span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
           throw se;
         }
@@ -198,8 +198,8 @@ final class JsonResumableSessionPutTask
         // a 503 with plain text content
         // Attempt to detect this very loosely as to minimize impact of modified error message
         // This is accurate circa 2023-06
-        if ((!JsonResumableSessionFailureScenario.isOk(code)
-                && !JsonResumableSessionFailureScenario.isContinue(code))
+        if ((!ResumableSessionFailureScenario.isOk(code)
+                && !ResumableSessionFailureScenario.isContinue(code))
             && contentType != null
             && contentType.startsWith("text/plain")
             && contentLength != null
@@ -207,14 +207,14 @@ final class JsonResumableSessionPutTask
           String errorMessage = cause.getContent().toLowerCase(Locale.US);
           if (errorMessage.contains("content-range")) {
             StorageException se =
-                JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+                ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
                     uploadId, response, cause, cause::getContent);
             span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
             throw se;
           }
         }
         StorageException se =
-            JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
+            ResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
         span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
         throw se;
       }
@@ -227,7 +227,7 @@ final class JsonResumableSessionPutTask
       throw e;
     } catch (Exception e) {
       StorageException se =
-          JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(uploadId, response, e);
+          ResumableSessionFailureScenario.SCENARIO_0.toStorageException(uploadId, response, e);
       span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
       throw se;
     } finally {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
index 5ce0de6f..57c5868c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java
@@ -55,7 +55,7 @@ final class JsonResumableSessionQueryTask
       response = req.execute();
 
       int code = response.getStatusCode();
-      if (JsonResumableSessionFailureScenario.isOk(code)) {
+      if (ResumableSessionFailureScenario.isOk(code)) {
         @Nullable StorageObject storageObject;
         @Nullable BigInteger actualSize;
 
@@ -74,7 +74,7 @@ final class JsonResumableSessionQueryTask
           storageObject = null;
         } else {
           response.ignore();
-          throw JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
+          throw ResumableSessionFailureScenario.SCENARIO_0_1.toStorageException(
               uploadId, response, null, () -> null);
         }
         if (actualSize != null) {
@@ -84,13 +84,13 @@ final class JsonResumableSessionQueryTask
             return ResumableOperationResult.incremental(actualSize.longValue());
           }
         } else {
-          throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+          throw ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
               uploadId,
               response,
               null,
               () -> storageObject != null ? storageObject.toString() : null);
         }
-      } else if (JsonResumableSessionFailureScenario.isContinue(code)) {
+      } else if (ResumableSessionFailureScenario.isContinue(code)) {
         String range1 = response.getHeaders().getRange();
         if (range1 != null) {
           ByteRangeSpec range = ByteRangeSpec.parse(range1);
@@ -110,23 +110,22 @@ final class JsonResumableSessionQueryTask
         // a 503 with plain text content
         // Attempt to detect this very loosely as to minimize impact of modified error message
         // This is accurate circa 2023-06
-        if ((!JsonResumableSessionFailureScenario.isOk(code)
-                && !JsonResumableSessionFailureScenario.isContinue(code))
+        if ((!ResumableSessionFailureScenario.isOk(code)
+                && !ResumableSessionFailureScenario.isContinue(code))
             && contentType != null
             && contentType.startsWith("text/plain")) {
           String errorMessage = cause.getContent().toLowerCase(Locale.US);
           if (errorMessage.contains("content-range")) {
-            throw JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException(
+            throw ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
                 uploadId, response, cause, cause::getContent);
           }
         }
-        throw JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
+        throw ResumableSessionFailureScenario.toStorageException(response, cause, uploadId);
       }
     } catch (StorageException se) {
       throw se;
     } catch (Exception e) {
-      throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
-          uploadId, response, e);
+      throw ResumableSessionFailureScenario.SCENARIO_0.toStorageException(uploadId, response, e);
     } finally {
       if (response != null) {
         try {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
similarity index 96%
rename from google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java
rename to google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
index 2e871178..0bccb2ff 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSessionFailureScenario.java
@@ -32,7 +32,7 @@ import javax.annotation.ParametersAreNonnullByDefault;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 @ParametersAreNonnullByDefault
-enum JsonResumableSessionFailureScenario {
+enum ResumableSessionFailureScenario {
   // TODO: send more bytes than are in the Content-Range header
   SCENARIO_0(BaseServiceException.UNKNOWN_CODE, null, "Unknown Error"),
   SCENARIO_0_1(BaseServiceException.UNKNOWN_CODE, null, "Response not application/json."),
@@ -87,7 +87,7 @@ enum JsonResumableSessionFailureScenario {
   @Nullable private final String reason;
   private final String message;
 
-  JsonResumableSessionFailureScenario(int code, @Nullable String reason, String message) {
+  ResumableSessionFailureScenario(int code, @Nullable String reason, String message) {
     this.code = code;
     this.reason = reason;
     this.message = message;
@@ -116,11 +116,15 @@ enum JsonResumableSessionFailureScenario {
     return toStorageException(code, message, reason, uploadId, resp, cause, contentCallable);
   }
 
+  StorageException toStorageException() {
+    return new StorageException(code, message, reason, null);
+  }
+
   static StorageException toStorageException(
       HttpResponse response, HttpResponseException cause, String uploadId) {
     String statusMessage = cause.getStatusMessage();
     StorageException se =
-        JsonResumableSessionFailureScenario.toStorageException(
+        ResumableSessionFailureScenario.toStorageException(
             cause.getStatusCode(),
             String.format(
                 "%d %s", cause.getStatusCode(), statusMessage == null ? "" : statusMessage),
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
index 75921032..b7bee985 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
@@ -84,12 +84,4 @@ final class ResumableWrite implements WriteObjectRequestBuilderFactory {
   public int hashCode() {
     return Objects.hash(req, res);
   }
-
-  /**
-   * Helper function which is more specific than {@link Function#identity()}. Constraining the input
-   * and output to be exactly {@link ResumableWrite}.
-   */
-  static ResumableWrite identity(ResumableWrite w) {
-    return w;
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableContent.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableContent.java
index 437ef49c..03d9cc46 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableContent.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableContent.java
@@ -61,6 +61,19 @@ abstract class RewindableContent extends AbstractHttpContent {
     return new ByteBufferContent(buffers);
   }
 
+  public static RewindableContent of(ByteBuffer[] srcs, int srcsOffset, int srcsLength) {
+    Preconditions.checkNotNull(srcs, "srcs must be non null");
+    if (!(0 <= srcsOffset && srcsOffset <= srcs.length)) {
+      throw new ArrayIndexOutOfBoundsException(
+          String.format(
+              "srcsOffset out of bounds (0 <= %d && %d <= %d)",
+              srcsOffset, srcsOffset, srcs.length));
+    }
+    Preconditions.checkArgument(srcsLength >= 0, "srcsLength >= 0 (%d >= 0)", srcsLength);
+    int end = srcsOffset + srcsLength;
+    return new ByteBufferContent(Arrays.copyOfRange(srcs, srcsOffset, end));
+  }
+
   static RewindableContent of(Path path) throws IOException {
     return new PathRewindableContent(path);
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
index 96fd933b..9bb6b7f5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java
@@ -20,7 +20,6 @@ import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
-import com.google.api.core.BetaApi;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
@@ -34,7 +33,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class DownloadJob {
 
   @NonNull private final List<ApiFuture<DownloadResult>> downloadResults;
@@ -55,7 +53,6 @@ public final class DownloadJob {
    *
    * @see Builder#setDownloadResults(List)
    */
-  @BetaApi
   public @NonNull List<DownloadResult> getDownloadResults() {
     return ApiExceptions.callAndTranslateApiException(ApiFutures.allAsList(downloadResults));
   }
@@ -65,7 +62,6 @@ public final class DownloadJob {
    *
    * @see Builder#setParallelDownloadConfig(ParallelDownloadConfig)
    */
-  @BetaApi
   public @NonNull ParallelDownloadConfig getParallelDownloadConfig() {
     return parallelDownloadConfig;
   }
@@ -96,7 +92,6 @@ public final class DownloadJob {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -106,7 +101,6 @@ public final class DownloadJob {
    *
    * @see DownloadJob
    */
-  @BetaApi
   public static final class Builder {
 
     private @NonNull List<ApiFuture<DownloadResult>> downloadResults;
@@ -122,7 +116,6 @@ public final class DownloadJob {
      * @return the instance of the Builder with DownloadResults modified.
      * @see DownloadJob#getDownloadResults()
      */
-    @BetaApi
     public Builder setDownloadResults(@NonNull List<ApiFuture<DownloadResult>> downloadResults) {
       this.downloadResults = ImmutableList.copyOf(downloadResults);
       return this;
@@ -134,7 +127,6 @@ public final class DownloadJob {
      * @return the instance of the Builder with ParallelDownloadConfig modified.
      * @see DownloadJob#getParallelDownloadConfig()
      */
-    @BetaApi
     public Builder setParallelDownloadConfig(
         @NonNull ParallelDownloadConfig parallelDownloadConfig) {
       this.parallelDownloadConfig = parallelDownloadConfig;
@@ -146,7 +138,6 @@ public final class DownloadJob {
      *
      * @return {@link DownloadJob}
      */
-    @BetaApi
     public DownloadJob build() {
       checkNotNull(downloadResults);
       checkNotNull(parallelDownloadConfig);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
index 0ac13bba..7084e755 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadResult.java
@@ -19,7 +19,6 @@ package com.google.cloud.storage.transfermanager;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.BlobInfo;
 import com.google.common.base.MoreObjects;
 import java.nio.file.Path;
@@ -33,7 +32,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class DownloadResult {
   static final Comparator<DownloadResult> COMPARATOR =
       Comparator.comparingInt(dr -> dr.getStatus().ordinal());
@@ -59,7 +57,6 @@ public final class DownloadResult {
    *
    * @see Builder#setInput(BlobInfo)
    */
-  @BetaApi
   public @NonNull BlobInfo getInput() {
     return input;
   }
@@ -70,7 +67,6 @@ public final class DownloadResult {
    *
    * @see Builder#setOutputDestination(Path)
    */
-  @BetaApi
   public @NonNull Path getOutputDestination() {
     checkState(
         status == TransferStatus.SUCCESS,
@@ -85,7 +81,6 @@ public final class DownloadResult {
    * @see TransferStatus
    * @see Builder#setStatus(TransferStatus)
    */
-  @BetaApi
   public @NonNull TransferStatus getStatus() {
     return status;
   }
@@ -97,7 +92,6 @@ public final class DownloadResult {
    *
    * @see Builder#setException(Exception)
    */
-  @BetaApi
   public @NonNull Exception getException() {
     checkState(
         status == TransferStatus.FAILED_TO_FINISH || status == TransferStatus.FAILED_TO_START,
@@ -136,7 +130,6 @@ public final class DownloadResult {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder(@NonNull BlobInfo blobInfo, @NonNull TransferStatus status) {
     return new Builder(blobInfo, status);
   }
@@ -146,7 +139,6 @@ public final class DownloadResult {
    *
    * @see DownloadResult
    */
-  @BetaApi
   public static final class Builder {
 
     private @NonNull BlobInfo input;
@@ -165,7 +157,6 @@ public final class DownloadResult {
      * @see DownloadResult#getInput()
      * @return the instance of the Builder with the value for input modified.
      */
-    @BetaApi
     public Builder setInput(@NonNull BlobInfo input) {
       this.input = input;
       return this;
@@ -178,7 +169,6 @@ public final class DownloadResult {
      * @see DownloadResult#getOutputDestination()
      * @return the instance of the Builder with the value for outputDestination modified.
      */
-    @BetaApi
     public Builder setOutputDestination(@NonNull Path outputDestination) {
       this.outputDestination = outputDestination;
       return this;
@@ -190,7 +180,6 @@ public final class DownloadResult {
      * @see TransferStatus
      * @return the instance of the Builder with the value for status modified.
      */
-    @BetaApi
     public Builder setStatus(@NonNull TransferStatus status) {
       this.status = status;
       return this;
@@ -204,7 +193,6 @@ public final class DownloadResult {
      * @see DownloadResult#getException()
      * @return the instance of the Builder with the value for exception modified.
      */
-    @BetaApi
     public Builder setException(@NonNull Exception exception) {
       this.exception = exception;
       return this;
@@ -215,7 +203,6 @@ public final class DownloadResult {
      *
      * @return {@link DownloadResult}
      */
-    @BetaApi
     public DownloadResult build() {
       checkNotNull(input);
       checkNotNull(status);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java
index 222dc9e3..0ad8bd91 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java
@@ -18,7 +18,6 @@ package com.google.cloud.storage.transfermanager;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
@@ -33,7 +32,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class ParallelDownloadConfig {
 
   @NonNull private final String stripPrefix;
@@ -57,7 +55,6 @@ public final class ParallelDownloadConfig {
    *
    * @see Builder#setStripPrefix(String)
    */
-  @BetaApi
   public @NonNull String getStripPrefix() {
     return stripPrefix;
   }
@@ -67,7 +64,6 @@ public final class ParallelDownloadConfig {
    *
    * @see Builder#setDownloadDirectory(Path)
    */
-  @BetaApi
   public @NonNull Path getDownloadDirectory() {
     return downloadDirectory;
   }
@@ -77,7 +73,6 @@ public final class ParallelDownloadConfig {
    *
    * @see Builder#setBucketName(String)
    */
-  @BetaApi
   public @NonNull String getBucketName() {
     return bucketName;
   }
@@ -88,7 +83,6 @@ public final class ParallelDownloadConfig {
    *
    * @see Builder#setOptionsPerRequest(List)
    */
-  @BetaApi
   public @NonNull List<BlobSourceOption> getOptionsPerRequest() {
     return optionsPerRequest;
   }
@@ -128,12 +122,10 @@ public final class ParallelDownloadConfig {
    *
    * @see ParallelDownloadConfig
    */
-  @BetaApi
   public static Builder newBuilder() {
     return new Builder();
   }
 
-  @BetaApi
   public static final class Builder {
 
     @NonNull private String stripPrefix;
@@ -155,7 +147,6 @@ public final class ParallelDownloadConfig {
      * @return the builder instance with the value for stripPrefix modified.
      * @see ParallelDownloadConfig#getStripPrefix()
      */
-    @BetaApi
     public Builder setStripPrefix(String stripPrefix) {
       this.stripPrefix = stripPrefix;
       return this;
@@ -167,7 +158,6 @@ public final class ParallelDownloadConfig {
      * @return the builder instance with the value for downloadDirectory modified.
      * @see ParallelDownloadConfig#getDownloadDirectory()
      */
-    @BetaApi
     public Builder setDownloadDirectory(Path downloadDirectory) {
       this.downloadDirectory = downloadDirectory;
       return this;
@@ -179,7 +169,6 @@ public final class ParallelDownloadConfig {
      * @return the builder instance with the value for bucketName modified.
      * @see ParallelDownloadConfig#getBucketName()
      */
-    @BetaApi
     public Builder setBucketName(String bucketName) {
       this.bucketName = bucketName;
       return this;
@@ -192,7 +181,6 @@ public final class ParallelDownloadConfig {
      * @return the builder instance with the value for OptionsPerRequest modified.
      * @see ParallelDownloadConfig#getOptionsPerRequest()
      */
-    @BetaApi
     public Builder setOptionsPerRequest(List<BlobSourceOption> optionsPerRequest) {
       this.optionsPerRequest = ImmutableList.copyOf(optionsPerRequest);
       return this;
@@ -203,7 +191,6 @@ public final class ParallelDownloadConfig {
      *
      * @return {@link ParallelDownloadConfig}
      */
-    @BetaApi
     public ParallelDownloadConfig build() {
       checkNotNull(bucketName);
       checkNotNull(stripPrefix);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
index b724d981..615d8542 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
@@ -18,7 +18,6 @@ package com.google.cloud.storage.transfermanager;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
@@ -31,7 +30,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class ParallelUploadConfig {
 
   private final boolean skipIfExists;
@@ -57,7 +55,6 @@ public final class ParallelUploadConfig {
    *
    * @see Builder#setSkipIfExists(boolean)
    */
-  @BetaApi
   public boolean isSkipIfExists() {
     return skipIfExists;
   }
@@ -67,7 +64,6 @@ public final class ParallelUploadConfig {
    *
    * @see Builder#setPrefix(String)
    */
-  @BetaApi
   public @NonNull String getPrefix() {
     return prefix;
   }
@@ -77,7 +73,6 @@ public final class ParallelUploadConfig {
    *
    * @see Builder#setBucketName(String)
    */
-  @BetaApi
   public @NonNull String getBucketName() {
     return bucketName;
   }
@@ -87,7 +82,6 @@ public final class ParallelUploadConfig {
    *
    * @see Builder#setWriteOptsPerRequest(List)
    */
-  @BetaApi
   public @NonNull List<BlobWriteOption> getWriteOptsPerRequest() {
     return writeOptsPerRequest;
   }
@@ -122,7 +116,6 @@ public final class ParallelUploadConfig {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -141,7 +134,6 @@ public final class ParallelUploadConfig {
    *
    * @see ParallelUploadConfig
    */
-  @BetaApi
   public static final class Builder {
 
     private boolean skipIfExists;
@@ -162,7 +154,6 @@ public final class ParallelUploadConfig {
      * @return the builder instance with the value for skipIfExists modified.
      * @see ParallelUploadConfig#isSkipIfExists()
      */
-    @BetaApi
     public Builder setSkipIfExists(boolean skipIfExists) {
       this.skipIfExists = skipIfExists;
       return this;
@@ -174,7 +165,6 @@ public final class ParallelUploadConfig {
      * @return the builder instance with the value for prefix modified.
      * @see ParallelUploadConfig#getPrefix()
      */
-    @BetaApi
     public Builder setPrefix(@NonNull String prefix) {
       this.prefix = prefix;
       return this;
@@ -186,7 +176,6 @@ public final class ParallelUploadConfig {
      * @return the builder instance with the value for bucketName modified.
      * @see ParallelUploadConfig#getBucketName()
      */
-    @BetaApi
     public Builder setBucketName(@NonNull String bucketName) {
       this.bucketName = bucketName;
       return this;
@@ -199,7 +188,6 @@ public final class ParallelUploadConfig {
      * @return the builder instance with the value for WriteOptsPerRequest modified.
      * @see ParallelUploadConfig#getWriteOptsPerRequest()
      */
-    @BetaApi
     public Builder setWriteOptsPerRequest(@NonNull List<BlobWriteOption> writeOptsPerRequest) {
       this.writeOptsPerRequest = writeOptsPerRequest;
       return this;
@@ -210,7 +198,6 @@ public final class ParallelUploadConfig {
      *
      * @return {@link ParallelUploadConfig}
      */
-    @BetaApi
     public ParallelUploadConfig build() {
       checkNotNull(prefix);
       checkNotNull(bucketName);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
index 9aae2345..d840523a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
@@ -16,7 +16,6 @@
 
 package com.google.cloud.storage.transfermanager;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.BlobInfo;
 import java.io.IOException;
 import java.nio.file.Path;
@@ -28,7 +27,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * <p>Transfer Manager handles Parallel Uploads and Parallel Downloads.
  */
-@BetaApi
 public interface TransferManager extends AutoCloseable {
 
   /**
@@ -57,7 +55,6 @@ public interface TransferManager extends AutoCloseable {
    *
    * @return an {@link UploadJob}
    */
-  @BetaApi
   @NonNull
   UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config) throws IOException;
 
@@ -89,7 +86,6 @@ public interface TransferManager extends AutoCloseable {
    *
    * @return a {@link DownloadJob}
    */
-  @BetaApi
   @NonNull
   DownloadJob downloadBlobs(List<BlobInfo> blobs, ParallelDownloadConfig config);
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
index f06839dd..3ffae091 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
@@ -16,7 +16,6 @@
 
 package com.google.cloud.storage.transfermanager;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.StorageOptions;
 import com.google.common.base.MoreObjects;
 import java.util.Objects;
@@ -26,7 +25,6 @@ import java.util.Objects;
  *
  * @see Builder
  */
-@BetaApi
 public final class TransferManagerConfig {
   private final int maxWorkers;
   private final int perWorkerBufferSize;
@@ -53,7 +51,6 @@ public final class TransferManagerConfig {
    *
    * @see Builder#setMaxWorkers(int)
    */
-  @BetaApi
   public int getMaxWorkers() {
     return maxWorkers;
   }
@@ -63,7 +60,6 @@ public final class TransferManagerConfig {
    *
    * @see Builder#setPerWorkerBufferSize(int)
    */
-  @BetaApi
   public int getPerWorkerBufferSize() {
     return perWorkerBufferSize;
   }
@@ -74,7 +70,6 @@ public final class TransferManagerConfig {
    *
    * @see Builder#setAllowDivideAndConquerDownload(boolean)
    */
-  @BetaApi
   public boolean isAllowDivideAndConquerDownload() {
     return allowDivideAndConquerDownload;
   }
@@ -93,7 +88,6 @@ public final class TransferManagerConfig {
    *     uploads</a> documentation for a more in depth explanation of the limitations of Parallel
    *     composite uploads.
    */
-  @BetaApi
   public boolean isAllowParallelCompositeUpload() {
     return allowParallelCompositeUpload;
   }
@@ -103,18 +97,15 @@ public final class TransferManagerConfig {
    *
    * @see Builder#setStorageOptions(StorageOptions)
    */
-  @BetaApi
   public StorageOptions getStorageOptions() {
     return storageOptions;
   }
 
   /** The service object for {@link TransferManager} */
-  @BetaApi
   public TransferManager getService() {
     return new TransferManagerImpl(this, DefaultQos.of(this));
   }
 
-  @BetaApi
   public Builder toBuilder() {
     return new Builder()
         .setAllowDivideAndConquerDownload(allowDivideAndConquerDownload)
@@ -161,7 +152,6 @@ public final class TransferManagerConfig {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -171,7 +161,6 @@ public final class TransferManagerConfig {
    *
    * @see TransferManagerConfig
    */
-  @BetaApi
   public static class Builder {
 
     private int maxWorkers;
@@ -198,7 +187,6 @@ public final class TransferManagerConfig {
      * @return the instance of Builder with the value for maxWorkers modified.
      * @see TransferManagerConfig#getMaxWorkers()
      */
-    @BetaApi
     public Builder setMaxWorkers(int maxWorkers) {
       this.maxWorkers = maxWorkers;
       return this;
@@ -212,7 +200,6 @@ public final class TransferManagerConfig {
      * @return the instance of Builder with the value for maxWorkers modified.
      * @see TransferManagerConfig#getPerWorkerBufferSize()
      */
-    @BetaApi
     public Builder setPerWorkerBufferSize(int perWorkerBufferSize) {
       this.perWorkerBufferSize = perWorkerBufferSize;
       return this;
@@ -227,7 +214,6 @@ public final class TransferManagerConfig {
      * @return the instance of Builder with the value for allowDivideAndConquerDownload modified.
      * @see TransferManagerConfig#isAllowDivideAndConquerDownload()
      */
-    @BetaApi
     public Builder setAllowDivideAndConquerDownload(boolean allowDivideAndConquerDownload) {
       this.allowDivideAndConquerDownload = allowDivideAndConquerDownload;
       return this;
@@ -242,7 +228,6 @@ public final class TransferManagerConfig {
      * @return the instance of Builder with the value for allowDivideAndConquerDownload modified.
      * @see TransferManagerConfig#isAllowDivideAndConquerDownload()
      */
-    @BetaApi
     public Builder setAllowParallelCompositeUpload(boolean allowParallelCompositeUpload) {
       this.allowParallelCompositeUpload = allowParallelCompositeUpload;
       return this;
@@ -256,7 +241,6 @@ public final class TransferManagerConfig {
      * @return the instance of Builder with the value for storageOptions modified.
      * @see TransferManagerConfig#getStorageOptions()
      */
-    @BetaApi
     public Builder setStorageOptions(StorageOptions storageOptions) {
       this.storageOptions = storageOptions;
       return this;
@@ -267,7 +251,6 @@ public final class TransferManagerConfig {
      *
      * @return {@link TransferManagerConfig}
      */
-    @BetaApi
     public TransferManagerConfig build() {
       return new TransferManagerConfig(
           maxWorkers,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
index 5489c1c8..6f3d8025 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
@@ -18,7 +18,6 @@ package com.google.cloud.storage.transfermanager;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
-import com.google.api.core.BetaApi;
 import com.google.api.core.ListenableFutureToApiFuture;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.FixedHeaderProvider;
@@ -48,7 +47,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.BinaryOperator;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
-@BetaApi
 final class TransferManagerImpl implements TransferManager {
 
   private static final String USER_AGENT_ENTRY = "gcloud-tm/";
@@ -103,7 +101,6 @@ final class TransferManagerImpl implements TransferManager {
   }
 
   @Override
-  @BetaApi
   public @NonNull UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config)
       throws IOException {
     Storage.BlobWriteOption[] opts =
@@ -134,7 +131,6 @@ final class TransferManagerImpl implements TransferManager {
   }
 
   @Override
-  @BetaApi
   public @NonNull DownloadJob downloadBlobs(List<BlobInfo> blobs, ParallelDownloadConfig config) {
     Storage.BlobSourceOption[] opts =
         config.getOptionsPerRequest().toArray(new Storage.BlobSourceOption[0]);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
index e3b5f4f1..31290e93 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferStatus.java
@@ -16,27 +16,21 @@
 
 package com.google.cloud.storage.transfermanager;
 
-import com.google.api.core.BetaApi;
 import java.util.Comparator;
 
 /** The status of a Upload/Download operation performed by Transfer Manager. */
-@BetaApi
 public enum TransferStatus {
   /** The transfer failed before bytes could be moved. */
-  @BetaApi
   FAILED_TO_START,
   /** The transfer failed after bytes could be moved. */
-  @BetaApi
   FAILED_TO_FINISH,
   /**
    * The transfer failed because the object/file already exists and skipIfExists was set to true.
    *
    * @see ParallelUploadConfig.Builder#setSkipIfExists(boolean)
    */
-  @BetaApi
   SKIPPED,
   /** The transfer was successful. */
-  @BetaApi
   SUCCESS;
 
   /** A null value is considered to be greater than all values */
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java
index 1b1050b0..cdaf9cf5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadJob.java
@@ -20,7 +20,6 @@ import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
-import com.google.api.core.BetaApi;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
@@ -34,7 +33,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class UploadJob {
 
   @NonNull private final List<ApiFuture<UploadResult>> uploadResults;
@@ -55,7 +53,6 @@ public final class UploadJob {
    *
    * @see Builder#setUploadResults(List)
    */
-  @BetaApi
   public List<UploadResult> getUploadResults() {
     return ApiExceptions.callAndTranslateApiException(ApiFutures.allAsList(uploadResults));
   }
@@ -65,7 +62,6 @@ public final class UploadJob {
    *
    * @see Builder#setParallelUploadConfig(ParallelUploadConfig)
    */
-  @BetaApi
   public ParallelUploadConfig getParallelUploadConfig() {
     return parallelUploadConfig;
   }
@@ -96,7 +92,6 @@ public final class UploadJob {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -106,7 +101,6 @@ public final class UploadJob {
    *
    * @see UploadJob
    */
-  @BetaApi
   public static final class Builder {
 
     private @NonNull List<ApiFuture<UploadResult>> uploadResults;
@@ -123,7 +117,6 @@ public final class UploadJob {
      * @return the instance of the Builder with UploadResults modified.
      * @see UploadJob#getUploadResults()
      */
-    @BetaApi
     public Builder setUploadResults(@NonNull List<ApiFuture<UploadResult>> uploadResults) {
       this.uploadResults = ImmutableList.copyOf(uploadResults);
       return this;
@@ -135,7 +128,6 @@ public final class UploadJob {
      * @return the instance of the Builder with ParallelUploadConfig modified.
      * @see UploadJob#getParallelUploadConfig()
      */
-    @BetaApi
     public Builder setParallelUploadConfig(@NonNull ParallelUploadConfig parallelUploadConfig) {
       this.parallelUploadConfig = parallelUploadConfig;
       return this;
@@ -146,7 +138,6 @@ public final class UploadJob {
      *
      * @return {@link UploadJob}
      */
-    @BetaApi
     public UploadJob build() {
       checkNotNull(uploadResults);
       checkNotNull(parallelUploadConfig);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java
index 3b9ba38f..fc83830b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadResult.java
@@ -19,7 +19,6 @@ package com.google.cloud.storage.transfermanager;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
-import com.google.api.core.BetaApi;
 import com.google.cloud.storage.BlobInfo;
 import com.google.common.base.MoreObjects;
 import java.util.Objects;
@@ -31,7 +30,6 @@ import org.checkerframework.checker.nullness.qual.NonNull;
  *
  * @see Builder
  */
-@BetaApi
 public final class UploadResult {
 
   @NonNull private final BlobInfo input;
@@ -55,7 +53,6 @@ public final class UploadResult {
    *
    * @see Builder#setInput(BlobInfo)
    */
-  @BetaApi
   public @NonNull BlobInfo getInput() {
     return input;
   }
@@ -66,7 +63,6 @@ public final class UploadResult {
    * @see TransferStatus
    * @see Builder#setStatus(TransferStatus)
    */
-  @BetaApi
   public @NonNull TransferStatus getStatus() {
     return status;
   }
@@ -77,7 +73,6 @@ public final class UploadResult {
    *
    * @see Builder#setUploadedBlob(BlobInfo)
    */
-  @BetaApi
   public @NonNull BlobInfo getUploadedBlob() {
     checkState(
         status == TransferStatus.SUCCESS,
@@ -93,7 +88,6 @@ public final class UploadResult {
    *
    * @see Builder#setException(Exception)
    */
-  @BetaApi
   public @NonNull Exception getException() {
     checkState(
         status == TransferStatus.FAILED_TO_START || status == TransferStatus.FAILED_TO_FINISH,
@@ -132,7 +126,6 @@ public final class UploadResult {
         .toString();
   }
 
-  @BetaApi
   public static Builder newBuilder(@NonNull BlobInfo input, @NonNull TransferStatus status) {
     return new Builder(input, status);
   }
@@ -142,7 +135,6 @@ public final class UploadResult {
    *
    * @see UploadResult
    */
-  @BetaApi
   public static final class Builder {
 
     private @NonNull BlobInfo input;
@@ -161,7 +153,6 @@ public final class UploadResult {
      * @return the Builder instance with the value for BlobInfo modified.
      * @see UploadResult#getInput()
      */
-    @BetaApi
     public Builder setInput(@NonNull BlobInfo input) {
       this.input = input;
       return this;
@@ -174,7 +165,6 @@ public final class UploadResult {
      * @see TransferStatus
      * @see UploadResult#getStatus()
      */
-    @BetaApi
     public Builder setStatus(@NonNull TransferStatus status) {
       this.status = status;
       return this;
@@ -186,7 +176,6 @@ public final class UploadResult {
      *
      * @return the Builder instance with the value for uploadedBlob modified.
      */
-    @BetaApi
     public Builder setUploadedBlob(@NonNull BlobInfo uploadedBlob) {
       this.uploadedBlob = uploadedBlob;
       return this;
@@ -200,7 +189,6 @@ public final class UploadResult {
      * @see UploadResult#getException()
      * @return the Builder instance with the value for exception modified.
      */
-    @BetaApi
     public Builder setException(@NonNull Exception exception) {
       this.exception = exception;
       return this;
@@ -211,7 +199,6 @@ public final class UploadResult {
      *
      * @return {@link UploadResult}
      */
-    @BetaApi
     public UploadResult build() {
       checkNotNull(input);
       checkNotNull(status);
diff --git a/grpc-google-cloud-storage-control-v2/pom.xml b/grpc-google-cloud-storage-control-v2/pom.xml
index 4567ef8e..87b56195 100644
--- a/grpc-google-cloud-storage-control-v2/pom.xml
+++ b/grpc-google-cloud-storage-control-v2/pom.xml
@@ -4,7 +4,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-  <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+  <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
   <name>grpc-google-cloud-storage-control-v2</name>
   <description>GRPC library for google-cloud-storage</description>
   <parent>
@@ -66,4 +66,4 @@
         </plugin>
     </plugins>
   </build>
-</project>
\ No newline at end of file
+</project>
diff --git a/pom.xml b/pom.xml
index d7c02eff..024837fe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -113,17 +113,17 @@
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>grpc-google-cloud-storage-control-v2</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.api.grpc</groupId>
         <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-storage-control</artifactId>
-        <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+        <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
diff --git a/proto-google-cloud-storage-control-v2/pom.xml b/proto-google-cloud-storage-control-v2/pom.xml
index 5a20bdb2..9e57bf1a 100644
--- a/proto-google-cloud-storage-control-v2/pom.xml
+++ b/proto-google-cloud-storage-control-v2/pom.xml
@@ -4,7 +4,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.api.grpc</groupId>
   <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-  <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
+  <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-storage-control-v2:current} -->
   <name>proto-google-cloud-storage-control-v2</name>
   <description>Proto library for  proto-google-cloud-storage-control-v2</description>
   <parent>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index f02de229..59ce2ff1 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -67,7 +67,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index b85c897a..050692d3 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -33,7 +33,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.37.1-alpha-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
+      <version>2.37.1-beta-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage-control:current} -->
       <scope>compile</scope>
     </dependency>
 
diff --git a/versions.txt b/versions.txt
index ade55818..efb19c1b 100644
--- a/versions.txt
+++ b/versions.txt
@@ -5,6 +5,6 @@ google-cloud-storage:2.37.0:2.37.1-SNAPSHOT
 gapic-google-cloud-storage-v2:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
 grpc-google-cloud-storage-v2:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
 proto-google-cloud-storage-v2:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
-google-cloud-storage-control:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
-proto-google-cloud-storage-control-v2:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
-grpc-google-cloud-storage-control-v2:2.37.0-alpha:2.37.1-alpha-SNAPSHOT
+google-cloud-storage-control:2.37.0-alpha:2.37.1-beta-SNAPSHOT
+proto-google-cloud-storage-control-v2:2.37.0-alpha:2.37.1-beta-SNAPSHOT
+grpc-google-cloud-storage-control-v2:2.37.0-alpha:2.37.1-beta-SNAPSHOT


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedChunkedResumableWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedChunkedResumableWritableByteChannelTest.java
new file mode 100644
index 00000000..3170afb7
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedChunkedResumableWritableByteChannelTest.java
@@ -0,0 +1,817 @@
+/*
+ * 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.ITGapicUnbufferedWritableByteChannelTest.DirectWriteService;
+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.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 com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import io.grpc.Status.Code;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.junit.Test;
+
+public final class ITGapicUnbufferedChunkedResumableWritableByteChannelTest {
+
+  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>
+   * WriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset= 0, checksummed_data.content.length = 262144 }
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ resources = {name = obj, size = 525288 } })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario1() throws Exception {
+
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .build();
+    WriteObjectResponse resp1 =
+        WriteObjectResponse.newBuilder()
+            .setResource(Object.newBuilder().setName("obj").setSize(_512KiB).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;
+
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      SettableApiFuture<WriteObjectResponse> done = SettableApiFuture.create();
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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>
+   * WriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 262144, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ persisted_size = 525288 })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario2() throws Exception {
+    String uploadId = "uploadId";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setFinishWrite(true)
+            .build();
+    WriteObjectResponse resp1 = WriteObjectResponse.newBuilder().setPersistedSize(_512KiB).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(_256KiB);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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>
+   * WriteObjectRequest{ upload_id = $UPLOAD_ID, write_offset = 524288, finish_write = true}
+   *     </pre></td>
+   *   </tr>
+   *   <tr>
+   *     <td>response</td>
+   *     <td><pre>
+   * onNext(WriteObjectResponse{ persisted_size = 262144 })
+   *     </pre></td>
+   *   </tr>
+   * </table>
+   */
+  @Test
+  public void scenario3() 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(_512KiB);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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>
+   * 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(_256KiB);
+
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              GrpcCallContext::createDefault);
+
+      channel.close();
+
+      WriteObjectResponse writeObjectResponse = done.get(2, TimeUnit.SECONDS);
+      assertThat(writeObjectResponse).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>
+   * 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(_512KiB);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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>
+   * 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(_512KiB);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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>
+   * WriteObjectRequest{ 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";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setWriteOffset(_256KiB)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB))))
+            .build();
+    StorageImplBase service1 =
+        new DirectWriteService(
+            (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<WriteObjectResponse> done = SettableApiFuture.create();
+      ResumableWrite resumableWrite = getResumableWrite(uploadId);
+      WriteCtx<ResumableWrite> writeCtx = new WriteCtx<>(resumableWrite);
+      writeCtx.getTotalSentBytes().set(_256KiB);
+      writeCtx.getConfirmedBytes().set(_256KiB);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .build();
+    WriteObjectResponse resp1 = WriteObjectResponse.newBuilder().setPersistedSize(_512KiB).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);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_256KiB)))
+                    .build())
+            .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);
+
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              CHUNK_SEGMENTER,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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";
+    WriteObjectRequest req1 =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(
+                ChecksummedData.newBuilder()
+                    .setContent(
+                        ByteString.copyFrom(DataGenerator.base64Characters().genBytes(_512KiB)))
+                    .build())
+            .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);
+
+      ChunkSegmenter chunkSegmenter =
+          new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), _512KiB, _256KiB);
+      //noinspection resource
+      GapicUnbufferedChunkedResumableWritableByteChannel channel =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              done,
+              chunkSegmenter,
+              storageClient.writeObjectCallable(),
+              writeCtx,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              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 ResumableWrite getResumableWrite(String uploadId) {
+    StartResumableWriteRequest req = StartResumableWriteRequest.getDefaultInstance();
+    StartResumableWriteResponse resp =
+        StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
+    return new ResumableWrite(
+        req, resp, id -> WriteObjectRequest.newBuilder().setUploadId(id).build());
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
index edbc8006..239b9bf3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
@@ -183,7 +183,7 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
               result,
               segmenter,
               sc.writeObjectCallable(),
-              reqFactory,
+              new WriteCtx<>(reqFactory),
               RetryingDependencies.attemptOnce(),
               Retrying.neverRetry(),
               Retrying::newCallContext);
@@ -266,7 +266,7 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
               result,
               segmenter,
               sc.writeObjectCallable(),
-              reqFactory,
+              new WriteCtx<>(reqFactory),
               TestUtils.defaultRetryingDeps(),
               new BasicResultRetryAlgorithm<Object>() {
                 @Override
@@ -323,7 +323,7 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
               result,
               segmenter,
               sc.writeObjectCallable(),
-              reqFactory,
+              new WriteCtx<>(reqFactory),
               RetryingDependencies.attemptOnce(),
               Retrying.neverRetry(),
               Retrying::newCallContext);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableSessionFailureScenarioTest.java
similarity index 91%
rename from google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java
rename to google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableSessionFailureScenarioTest.java
index f40cf27f..0a53a4bb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableSessionFailureScenarioTest.java
@@ -16,8 +16,8 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isContinue;
-import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isOk;
+import static com.google.cloud.storage.ResumableSessionFailureScenario.isContinue;
+import static com.google.cloud.storage.ResumableSessionFailureScenario.isOk;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.client.http.EmptyContent;
@@ -33,7 +33,7 @@ import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
 import org.junit.Test;
 
-public final class JsonResumableSessionFailureScenarioTest {
+public final class ResumableSessionFailureScenarioTest {
   private static final GsonFactory gson = GsonFactory.getDefaultInstance();
 
   @Test
@@ -64,7 +64,7 @@ public final class JsonResumableSessionFailureScenarioTest {
     resp.getHeaders().setContentType("text/plain; charset=utf-8").setContentLength(5L);
 
     StorageException storageException =
-        JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException(
+        ResumableSessionFailureScenario.SCENARIO_1.toStorageException(
             "uploadId",
             resp,
             new Cause(),
@@ -105,7 +105,7 @@ public final class JsonResumableSessionFailureScenarioTest {
         .setContentLength((long) bytes.length);
 
     StorageException storageException =
-        JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+        ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
             "uploadId", resp, null, () -> json);
 
     assertThat(storageException.getCode()).isEqualTo(0);
@@ -128,7 +128,7 @@ public final class JsonResumableSessionFailureScenarioTest {
         .setContentLength(0L);
 
     StorageException storageException =
-        JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+        ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
             "uploadId", resp, null, () -> null);
 
     assertThat(storageException.getCode()).isEqualTo(0);
@@ -151,7 +151,7 @@ public final class JsonResumableSessionFailureScenarioTest {
     resp.getHeaders().set("X-Goog-Gcs-Idempotency-Token", "5").setContentLength(0L);
 
     StorageException storageException =
-        JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(
+        ResumableSessionFailureScenario.SCENARIO_0.toStorageException(
             "uploadId", resp, null, () -> null);
 
     assertThat(storageException.getCode()).isEqualTo(0);
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 27e925f8..e1eb11c9 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
@@ -103,9 +103,9 @@ public final class TestUtils {
   }
 
   public static ApiException apiException(Code code, String message) {
-    StatusRuntimeException statusRuntimeException = code.toStatus().asRuntimeException();
-    DebugInfo debugInfo =
-        DebugInfo.newBuilder().setDetail("forced failure |~| " + code.name() + message).build();
+    StatusRuntimeException statusRuntimeException =
+        code.toStatus().withDescription(message).asRuntimeException();
+    DebugInfo debugInfo = DebugInfo.newBuilder().setDetail(message).build();
     ErrorDetails errorDetails =
         ErrorDetails.builder().setRawErrorMessages(ImmutableList.of(Any.pack(debugInfo))).build();
     return ApiExceptionFactory.createException(