Repository: googleapis/java-storage
Pull request #2514 (2024-04-24 19:21:54)

chore: remove WriteFlusher instead having dedicated WritableByteChannels

WriteFlusher has grown unwieldy to be able to properly handle the kinds of uploads that are performed while being able to perform appropriate fine grained response validation.

This is the first change in a series to remove WriteFlusher and create purpose dedicated UnbufferedWritableByteChannels for the different kinds of uploads we perform. By having dedicated channels more fine grained response validation and request packing can be performed.


I left multiple commits in this PR for the time being. They are generally:
1. Introduce GapicUnbufferedDirectWritableByteChannel used for direct/single-shot uploads
2. Introduce GapicUnbufferedChunkedResumableWritableByteChannel used for chunked resumable uploads
3. Introduce GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel used for for resumable uploads from files on disk
4. Refactor GapicBidiUnbufferedWritableByteChannel
5. Cleanup and removal of the majority of WriteFlushStrategy.java and its corresponding tests now that the functionality is distributed in each of the dedicated channel implementations


Patch:
diff --git a/README.md b/README.md
index 4cabaedc..d5706ef3 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.37.0</version>
+      <version>26.38.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -46,7 +46,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>2.36.1</version>
+  <version>2.37.0</version>
 </dependency>
 <dependency>
   <groupId>com.google.cloud</groupId>
@@ -59,7 +59,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.37.0')
+implementation platform('com.google.cloud:libraries-bom:26.38.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
diff --git a/gapic-google-cloud-storage-v2/src/main/java/com/google/storage/v2/stub/StorageStubSettings.java b/gapic-google-cloud-storage-v2/src/main/java/com/google/storage/v2/stub/StorageStubSettings.java
index e4787eb0..1b42b3c7 100644
--- a/gapic-google-cloud-storage-v2/src/main/java/com/google/storage/v2/stub/StorageStubSettings.java
+++ b/gapic-google-cloud-storage-v2/src/main/java/com/google/storage/v2/stub/StorageStubSettings.java
@@ -625,15 +625,6 @@ public class StorageStubSettings extends StubSettings<StorageStubSettings> {
             "Transport not supported: %s", getTransportChannelProvider().getTransportName()));
   }
 
-  /** Returns the endpoint set by the user or the the service's default endpoint. */
-  @Override
-  public String getEndpoint() {
-    if (super.getEndpoint() != null) {
-      return super.getEndpoint();
-    }
-    return getDefaultEndpoint();
-  }
-
   /** Returns the default service name. */
   @Override
   public String getServiceName() {
@@ -1340,15 +1331,6 @@ public class StorageStubSettings extends StubSettings<StorageStubSettings> {
       return updateHmacKeySettings;
     }
 
-    /** Returns the endpoint set by the user or the the service's default endpoint. */
-    @Override
-    public String getEndpoint() {
-      if (super.getEndpoint() != null) {
-        return super.getEndpoint();
-      }
-      return getDefaultEndpoint();
-    }
-
     @Override
     public StorageStubSettings build() throws IOException {
       return new StorageStubSettings(this);
diff --git a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/StorageControlClient.java b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/StorageControlClient.java
index a25fbd67..49310103 100644
--- a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/StorageControlClient.java
+++ b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/StorageControlClient.java
@@ -70,7 +70,7 @@ import javax.annotation.Generated;
  *    </tr>
  *    <tr>
  *      <td><p> CreateFolder</td>
- *      <td><p> Creates a new folder.</td>
+ *      <td><p> Creates a new folder. This operation is only applicable to a hierarchical namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.</td>
  *      <td>
  *      <p>Request object method variants only take one parameter, a request object, which must be constructed before the call.</p>
  *      <ul>
@@ -89,7 +89,7 @@ import javax.annotation.Generated;
  *    </tr>
  *    <tr>
  *      <td><p> DeleteFolder</td>
- *      <td><p> Permanently deletes an empty folder.</td>
+ *      <td><p> Permanently deletes an empty folder. This operation is only applicable to a hierarchical namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.</td>
  *      <td>
  *      <p>Request object method variants only take one parameter, a request object, which must be constructed before the call.</p>
  *      <ul>
@@ -108,7 +108,7 @@ import javax.annotation.Generated;
  *    </tr>
  *    <tr>
  *      <td><p> GetFolder</td>
- *      <td><p> Returns metadata for the specified folder.</td>
+ *      <td><p> Returns metadata for the specified folder. This operation is only applicable to a hierarchical namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.</td>
  *      <td>
  *      <p>Request object method variants only take one parameter, a request object, which must be constructed before the call.</p>
  *      <ul>
@@ -127,7 +127,7 @@ import javax.annotation.Generated;
  *    </tr>
  *    <tr>
  *      <td><p> ListFolders</td>
- *      <td><p> Retrieves a list of folders for a given bucket.</td>
+ *      <td><p> Retrieves a list of folders. This operation is only applicable to a hierarchical namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.</td>
  *      <td>
  *      <p>Request object method variants only take one parameter, a request object, which must be constructed before the call.</p>
  *      <ul>
@@ -147,7 +147,7 @@ import javax.annotation.Generated;
  *    </tr>
  *    <tr>
  *      <td><p> RenameFolder</td>
- *      <td><p> Renames a source folder to a destination folder. During a rename, the source and destination folders are locked until the long running operation completes.</td>
+ *      <td><p> Renames a source folder to a destination folder. This operation is only applicable to a hierarchical namespace enabled bucket. During a rename, the source and destination folders are locked until the long running operation completes. Hierarchical namespace buckets are in allowlist preview.</td>
  *      <td>
  *      <p>Request object method variants only take one parameter, a request object, which must be constructed before the call.</p>
  *      <ul>
@@ -365,7 +365,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Creates a new folder.
+   * Creates a new folder. This operation is only applicable to a hierarchical namespace enabled
+   * bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -383,7 +384,8 @@ public class StorageControlClient implements BackgroundResource {
    * }
    * }</pre>
    *
-   * @param parent Required. Name of the bucket in which the folder will reside.
+   * @param parent Required. Name of the bucket in which the folder will reside. The bucket must be
+   *     a hierarchical namespace enabled bucket.
    * @param folder Required. Properties of the new folder being created. The bucket and name of the
    *     folder are specified in the parent and folder_id fields, respectively. Populating those
    *     fields in `folder` will result in an error.
@@ -405,7 +407,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Creates a new folder.
+   * Creates a new folder. This operation is only applicable to a hierarchical namespace enabled
+   * bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -423,7 +426,8 @@ public class StorageControlClient implements BackgroundResource {
    * }
    * }</pre>
    *
-   * @param parent Required. Name of the bucket in which the folder will reside.
+   * @param parent Required. Name of the bucket in which the folder will reside. The bucket must be
+   *     a hierarchical namespace enabled bucket.
    * @param folder Required. Properties of the new folder being created. The bucket and name of the
    *     folder are specified in the parent and folder_id fields, respectively. Populating those
    *     fields in `folder` will result in an error.
@@ -445,7 +449,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Creates a new folder.
+   * Creates a new folder. This operation is only applicable to a hierarchical namespace enabled
+   * bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -477,7 +482,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Creates a new folder.
+   * Creates a new folder. This operation is only applicable to a hierarchical namespace enabled
+   * bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -508,7 +514,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Permanently deletes an empty folder.
+   * Permanently deletes an empty folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -536,7 +543,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Permanently deletes an empty folder.
+   * Permanently deletes an empty folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -563,7 +571,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Permanently deletes an empty folder.
+   * Permanently deletes an empty folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -594,7 +603,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Permanently deletes an empty folder.
+   * Permanently deletes an empty folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -624,7 +634,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Returns metadata for the specified folder.
+   * Returns metadata for the specified folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -652,7 +663,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Returns metadata for the specified folder.
+   * Returns metadata for the specified folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -679,7 +691,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Returns metadata for the specified folder.
+   * Returns metadata for the specified folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -710,7 +723,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Returns metadata for the specified folder.
+   * Returns metadata for the specified folder. This operation is only applicable to a hierarchical
+   * namespace enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -740,7 +754,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Retrieves a list of folders for a given bucket.
+   * Retrieves a list of folders. This operation is only applicable to a hierarchical namespace
+   * enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -758,7 +773,8 @@ public class StorageControlClient implements BackgroundResource {
    * }
    * }</pre>
    *
-   * @param parent Required. Name of the bucket in which to look for folders.
+   * @param parent Required. Name of the bucket in which to look for folders. The bucket must be a
+   *     hierarchical namespace enabled bucket.
    * @throws com.google.api.gax.rpc.ApiException if the remote call fails
    */
   public final ListFoldersPagedResponse listFolders(BucketName parent) {
@@ -771,7 +787,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Retrieves a list of folders for a given bucket.
+   * Retrieves a list of folders. This operation is only applicable to a hierarchical namespace
+   * enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -789,7 +806,8 @@ public class StorageControlClient implements BackgroundResource {
    * }
    * }</pre>
    *
-   * @param parent Required. Name of the bucket in which to look for folders.
+   * @param parent Required. Name of the bucket in which to look for folders. The bucket must be a
+   *     hierarchical namespace enabled bucket.
    * @throws com.google.api.gax.rpc.ApiException if the remote call fails
    */
   public final ListFoldersPagedResponse listFolders(String parent) {
@@ -799,7 +817,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Retrieves a list of folders for a given bucket.
+   * Retrieves a list of folders. This operation is only applicable to a hierarchical namespace
+   * enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -836,7 +855,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Retrieves a list of folders for a given bucket.
+   * Retrieves a list of folders. This operation is only applicable to a hierarchical namespace
+   * enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -874,7 +894,8 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Retrieves a list of folders for a given bucket.
+   * Retrieves a list of folders. This operation is only applicable to a hierarchical namespace
+   * enabled bucket. Hierarchical namespace buckets are in allowlist preview.
    *
    * <p>Sample code:
    *
@@ -917,8 +938,10 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Renames a source folder to a destination folder. During a rename, the source and destination
-   * folders are locked until the long running operation completes.
+   * Renames a source folder to a destination folder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket. During a rename, the source and destination folders are
+   * locked until the long running operation completes. Hierarchical namespace buckets are in
+   * allowlist preview.
    *
    * <p>Sample code:
    *
@@ -952,8 +975,10 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Renames a source folder to a destination folder. During a rename, the source and destination
-   * folders are locked until the long running operation completes.
+   * Renames a source folder to a destination folder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket. During a rename, the source and destination folders are
+   * locked until the long running operation completes. Hierarchical namespace buckets are in
+   * allowlist preview.
    *
    * <p>Sample code:
    *
@@ -987,8 +1012,10 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Renames a source folder to a destination folder. During a rename, the source and destination
-   * folders are locked until the long running operation completes.
+   * Renames a source folder to a destination folder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket. During a rename, the source and destination folders are
+   * locked until the long running operation completes. Hierarchical namespace buckets are in
+   * allowlist preview.
    *
    * <p>Sample code:
    *
@@ -1021,8 +1048,10 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Renames a source folder to a destination folder. During a rename, the source and destination
-   * folders are locked until the long running operation completes.
+   * Renames a source folder to a destination folder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket. During a rename, the source and destination folders are
+   * locked until the long running operation completes. Hierarchical namespace buckets are in
+   * allowlist preview.
    *
    * <p>Sample code:
    *
@@ -1055,8 +1084,10 @@ public class StorageControlClient implements BackgroundResource {
 
   // AUTO-GENERATED DOCUMENTATION AND METHOD.
   /**
-   * Renames a source folder to a destination folder. During a rename, the source and destination
-   * folders are locked until the long running operation completes.
+   * Renames a source folder to a destination folder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket. During a rename, the source and destination folders are
+   * locked until the long running operation completes. Hierarchical namespace buckets are in
+   * allowlist preview.
    *
    * <p>Sample code:
    *
diff --git a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/package-info.java b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/package-info.java
index 126dd95d..4812aba8 100644
--- a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/package-info.java
+++ b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/package-info.java
@@ -15,7 +15,7 @@
  */
 
 /**
- * A client to Cloud Storage API
+ * A client to Storage Control API
  *
  * <p>The interfaces provided are listed below, along with usage samples.
  *
diff --git a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/stub/StorageControlStubSettings.java b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/stub/StorageControlStubSettings.java
index 5b34d700..eb717d38 100644
--- a/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/stub/StorageControlStubSettings.java
+++ b/google-cloud-storage-control/src/main/java/com/google/storage/control/v2/stub/StorageControlStubSettings.java
@@ -322,15 +322,6 @@ public class StorageControlStubSettings extends StubSettings<StorageControlStubS
             "Transport not supported: %s", getTransportChannelProvider().getTransportName()));
   }
 
-  /** Returns the endpoint set by the user or the the service's default endpoint. */
-  @Override
-  public String getEndpoint() {
-    if (super.getEndpoint() != null) {
-      return super.getEndpoint();
-    }
-    return getDefaultEndpoint();
-  }
-
   /** Returns the default service name. */
   @Override
   public String getServiceName() {
@@ -715,15 +706,6 @@ public class StorageControlStubSettings extends StubSettings<StorageControlStubS
       return listManagedFoldersSettings;
     }
 
-    /** Returns the endpoint set by the user or the the service's default endpoint. */
-    @Override
-    public String getEndpoint() {
-      if (super.getEndpoint() != null) {
-        return super.getEndpoint();
-      }
-      return getDefaultEndpoint();
-    }
-
     @Override
     public StorageControlStubSettings build() throws IOException {
       return new StorageControlStubSettings(this);
diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index cc9d6933..df4e590c 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -78,4 +78,10 @@
     <method>com.google.cloud.storage.StorageOptions$Builder setBlobWriteSessionConfig(com.google.cloud.storage.BlobWriteSessionConfig)</method>
   </difference>
 
+  <!-- somehow clirr things a public class in a package-private class is part of the public api -->
+  <difference>
+    <differenceType>8001</differenceType>
+    <className>com/google/cloud/storage/WriteFlushStrategy$DefaultBidiFlusher</className>
+  </difference>
+
 </differences>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 225fa959..4528c5b8 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -16,7 +16,7 @@
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <pubsub-proto.version>1.111.0</pubsub-proto.version>
+    <pubsub-proto.version>1.111.2</pubsub-proto.version>
     <junit-platform.version>5.10.2</junit-platform.version>
   </properties>
   <dependencies>
@@ -181,13 +181,13 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-kms-v1</artifactId>
-      <version>0.135.0</version>
+      <version>0.136.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-kms</artifactId>
-      <version>2.44.0</version>
+      <version>2.45.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
index d19a8802..19aba735 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
@@ -16,9 +16,17 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
+
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.BidiStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
+import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
@@ -31,33 +39,45 @@ import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.function.Supplier;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
-final class GapicBidiUnbufferedWritableByteChannel<
-        RequestFactoryT extends BidiWriteCtx.BidiWriteObjectRequestBuilderFactory>
-    implements UnbufferedWritableByteChannel {
-
+final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
+  private final BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write;
+  private final RetryingDependencies deps;
+  private final ResultRetryAlgorithm<?> alg;
+  private final String bucketName;
+  private final Supplier<GrpcCallContext> baseContextSupplier;
   private final SettableApiFuture<BidiWriteObjectResponse> resultFuture;
   private final ChunkSegmenter chunkSegmenter;
 
-  private final BidiWriteCtx<RequestFactoryT> writeCtx;
-  private final WriteFlushStrategy.BidiFlusher flusher;
+  private final BidiWriteCtx<BidiResumableWrite> writeCtx;
+  private final BidiObserver responseObserver;
 
+  private volatile ApiStreamObserver<BidiWriteObjectRequest> stream;
   private boolean open = true;
+  private boolean first = true;
   private boolean finished = false;
 
   GapicBidiUnbufferedWritableByteChannel(
+      BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg,
       SettableApiFuture<BidiWriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
-      RequestFactoryT requestFactory,
-      WriteFlushStrategy.BidiFlusherFactory flusherFactory) {
+      BidiResumableWrite requestFactory,
+      Supplier<GrpcCallContext> baseContextSupplier) {
+    this.write = write;
+    this.deps = deps;
+    this.alg = alg;
+    this.baseContextSupplier = baseContextSupplier;
+    this.bucketName = requestFactory.bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
     this.writeCtx = new BidiWriteCtx<>(requestFactory);
-    this.flusher =
-        flusherFactory.newFlusher(
-            requestFactory.bucketName(), writeCtx.getConfirmedBytes()::set, resultFuture::set);
+    this.responseObserver = new BidiObserver();
   }
 
   @Override
@@ -65,6 +85,13 @@ final class GapicBidiUnbufferedWritableByteChannel<
     return internalWrite(srcs, srcsOffset, srcsLength, false);
   }
 
+  @Override
+  public long writeAndClose(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    long written = internalWrite(srcs, offset, length, true);
+    close();
+    return written;
+  }
+
   @Override
   public boolean isOpen() {
     return open;
@@ -72,23 +99,29 @@ final class GapicBidiUnbufferedWritableByteChannel<
 
   @Override
   public void close() throws IOException {
+    if (!open) {
+      return;
+    }
+    ApiStreamObserver<BidiWriteObjectRequest> openedStream = openedStream();
     if (!finished) {
       BidiWriteObjectRequest message = finishMessage();
       try {
-        flusher.close(message);
+        openedStream.onNext(message);
         finished = true;
+        openedStream.onCompleted();
       } catch (RuntimeException e) {
         resultFuture.setException(e);
         throw e;
       }
     } else {
-      flusher.close(null);
+      openedStream.onCompleted();
     }
+    responseObserver.await();
     open = false;
   }
 
   @VisibleForTesting
-  BidiWriteCtx<RequestFactoryT> getWriteCtx() {
+  BidiWriteCtx<BidiResumableWrite> getWriteCtx() {
     return writeCtx;
   }
 
@@ -130,7 +163,8 @@ final class GapicBidiUnbufferedWritableByteChannel<
         finished = true;
       }
 
-      BidiWriteObjectRequest build = builder.build();
+      BidiWriteObjectRequest build = possiblyPairDownBidiRequest(builder, first).build();
+      first = false;
       messages.add(build);
       bytesConsumed += contentSize;
     }
@@ -140,7 +174,7 @@ final class GapicBidiUnbufferedWritableByteChannel<
     }
 
     try {
-      flusher.flush(messages);
+      flush(messages);
     } catch (RuntimeException e) {
       resultFuture.setException(e);
       throw e;
@@ -162,4 +196,123 @@ final class GapicBidiUnbufferedWritableByteChannel<
     BidiWriteObjectRequest message = b.build();
     return message;
   }
+
+  private ApiStreamObserver<BidiWriteObjectRequest> openedStream() {
+    if (stream == null) {
+      synchronized (this) {
+        if (stream == null) {
+          GrpcCallContext internalContext =
+              contextWithBucketName(bucketName, baseContextSupplier.get());
+          stream =
+              this.write
+                  .withDefaultCallContext(internalContext)
+                  .bidiStreamingCall(responseObserver);
+          responseObserver.sem.drainPermits();
+        }
+      }
+    }
+    return stream;
+  }
+
+  private void flush(@NonNull List<BidiWriteObjectRequest> segments) {
+    Retrying.run(
+        deps,
+        alg,
+        () -> {
+          try {
+            ApiStreamObserver<BidiWriteObjectRequest> opened = openedStream();
+            for (BidiWriteObjectRequest message : segments) {
+              opened.onNext(message);
+            }
+            if (!finished) {
+              BidiWriteObjectRequest message =
+                  BidiWriteObjectRequest.newBuilder().setFlush(true).setStateLookup(true).build();
+              opened.onNext(message);
+            }
+            responseObserver.await();
+            return null;
+          } catch (Exception e) {
+            stream = null;
+            first = true;
+            throw e;
+          }
+        },
+        Decoder.identity());
+  }
+
+  private static BidiWriteObjectRequest.Builder possiblyPairDownBidiRequest(
+      BidiWriteObjectRequest.Builder b, boolean firstMessageOfStream) {
+    if (firstMessageOfStream && b.getWriteOffset() == 0) {
+      return b;
+    }
+
+    if (!firstMessageOfStream) {
+      b.clearUploadId();
+    }
+
+    if (b.getWriteOffset() > 0) {
+      b.clearWriteObjectSpec();
+    }
+
+    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
+      b.clearObjectChecksums();
+    }
+    return b;
+  }
+
+  private class BidiObserver implements ApiStreamObserver<BidiWriteObjectResponse> {
+
+    private final Semaphore sem;
+    private volatile BidiWriteObjectResponse last;
+    private volatile RuntimeException previousError;
+
+    private BidiObserver() {
+      this.sem = new Semaphore(0);
+    }
+
+    @Override
+    public void onNext(BidiWriteObjectResponse value) {
+      // incremental update
+      if (value.hasPersistedSize()) {
+        writeCtx.getConfirmedBytes().set((value.getPersistedSize()));
+      } else if (value.hasResource()) {
+        writeCtx.getConfirmedBytes().set(value.getResource().getSize());
+      }
+      sem.release();
+      last = value;
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      if (t instanceof RuntimeException) {
+        previousError = (RuntimeException) t;
+      }
+      sem.release();
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        resultFuture.set(last);
+      }
+      sem.release();
+    }
+
+    void await() {
+      try {
+        sem.acquire();
+      } catch (InterruptedException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        } else {
+          throw new RuntimeException(e);
+        }
+      }
+      RuntimeException err = previousError;
+      if (err != null) {
+        previousError = null;
+        throw err;
+      }
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
index e26d33bb..536eba7f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
@@ -30,7 +30,6 @@ import com.google.storage.v2.BidiWriteObjectResponse;
 import com.google.storage.v2.ServiceConstants.Values;
 import java.nio.ByteBuffer;
 import java.util.function.BiFunction;
-import java.util.function.Function;
 
 final class GapicBidiWritableByteChannelSessionBuilder {
 
@@ -78,36 +77,6 @@ final class GapicBidiWritableByteChannelSessionBuilder {
     return this;
   }
 
-  /**
-   * When constructing a bidi channel session, there is always a {@link
-   * GapicBidiUnbufferedWritableByteChannel} at the bottom of it. This method creates a BiFunction
-   * which will instantiate the {@link GapicBidiUnbufferedWritableByteChannel} when provided with a
-   * {@code StartT} value and a {@code SettableApiFuture<BidiWriteObjectResponse>}.
-   *
-   * <p>As part of providing the function, the provided parameters {@code BidiFlusherFactory} and
-   * {@code f} are "bound" into the returned function. In conjunction with the configured fields of
-   * this class a new instance of {@link GapicBidiUnbufferedWritableByteChannel} can be constructed.
-   */
-  private <StartT, RequestFactoryT extends BidiWriteCtx.BidiWriteObjectRequestBuilderFactory>
-      BiFunction<StartT, SettableApiFuture<BidiWriteObjectResponse>, UnbufferedWritableByteChannel>
-          bindFunction(
-              WriteFlushStrategy.BidiFlusherFactory flusherFactory,
-              Function<StartT, RequestFactoryT> f) {
-    // it is theoretically possible that the setter methods for the following variables could
-    // be called again between when this method is invoked and the resulting function is invoked.
-    // To ensure we are using the specified values at the point in time they are bound to the
-    // function read them into local variables which will be closed over rather than the class
-    // fields.
-    ByteStringStrategy boundStrategy = byteStringStrategy;
-    Hasher boundHasher = hasher;
-    return (start, resultFuture) ->
-        new GapicBidiUnbufferedWritableByteChannel<>(
-            resultFuture,
-            new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-            f.apply(start),
-            flusherFactory);
-  }
-
   GapicBidiWritableByteChannelSessionBuilder.ResumableUploadBuilder resumable() {
     return new GapicBidiWritableByteChannelSessionBuilder.ResumableUploadBuilder();
   }
@@ -164,12 +133,30 @@ final class GapicBidiWritableByteChannelSessionBuilder {
       }
 
       BufferedWritableByteChannelSession<BidiWriteObjectResponse> build() {
+        // it is theoretically possible that the setter methods for the following variables could
+        // be called again between when this method is invoked and the resulting function is
+        // invoked.
+        // To ensure we are using the specified values at the point in time they are bound to the
+        // function read them into local variables which will be closed over rather than the class
+        // fields.
+        ByteStringStrategy boundStrategy = byteStringStrategy;
+        Hasher boundHasher = hasher;
         return new BufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
-            bindFunction(
-                    WriteFlushStrategy.defaultBidiFlusher(
-                        write, deps, alg, Retrying::newCallContext),
-                    BidiResumableWrite::identity)
+            ((BiFunction<
+                        BidiResumableWrite,
+                        SettableApiFuture<BidiWriteObjectResponse>,
+                        UnbufferedWritableByteChannel>)
+                    (start, resultFuture) ->
+                        new GapicBidiUnbufferedWritableByteChannel(
+                            write,
+                            deps,
+                            alg,
+                            resultFuture,
+                            new ChunkSegmenter(
+                                boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
+                            start,
+                            Retrying::newCallContext))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
new file mode 100644
index 00000000..aa48810d
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java
@@ -0,0 +1,303 @@
+/*
+ * Copyright 2022 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.GrpcUtils.contextWithBucketName;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.ObjectChecksums;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import java.util.function.LongConsumer;
+import java.util.function.Supplier;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class GapicUnbufferedChunkedResumableWritableByteChannel
+    implements UnbufferedWritableByteChannel {
+
+  private final SettableApiFuture<WriteObjectResponse> resultFuture;
+  private final ChunkSegmenter chunkSegmenter;
+  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+
+  private final String bucketName;
+  private final WriteCtx<ResumableWrite> writeCtx;
+  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 boolean finished = false;
+
+  GapicUnbufferedChunkedResumableWritableByteChannel(
+      SettableApiFuture<WriteObjectResponse> resultFuture,
+      @NonNull ChunkSegmenter chunkSegmenter,
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      ResumableWrite requestFactory,
+      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.deps = deps;
+    this.alg = alg;
+    this.baseContextSupplier = baseContextSupplier;
+    this.sizeCallback = writeCtx.getConfirmedBytes()::set;
+    this.completeCallback = resultFuture::set;
+  }
+
+  @Override
+  public long write(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    return internalWrite(srcs, srcsOffset, srcsLength, false);
+  }
+
+  @Override
+  public long writeAndClose(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    long write = internalWrite(srcs, srcsOffset, srcsLength, true);
+    close();
+    return write;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (open && !finished) {
+      WriteObjectRequest message = finishMessage(true);
+      try {
+        flush(ImmutableList.of(message));
+        finished = true;
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
+    }
+    open = false;
+  }
+
+  private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
+      throws ClosedChannelException {
+    if (!open) {
+      throw new ClosedChannelException();
+    }
+
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+
+    List<WriteObjectRequest> messages = new ArrayList<>();
+
+    boolean first = true;
+    int bytesConsumed = 0;
+    for (ChunkSegment datum : data) {
+      Crc32cLengthKnown crc32c = datum.getCrc32c();
+      ByteString b = datum.getB();
+      int contentSize = b.size();
+      long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+      Crc32cLengthKnown cumulative =
+          writeCtx
+              .getCumulativeCrc32c()
+              .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
+      ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
+      if (crc32c != null) {
+        checksummedData.setCrc32C(crc32c.getValue());
+      }
+      WriteObjectRequest.Builder builder =
+          writeCtx
+              .newRequestBuilder()
+              .setWriteOffset(offset)
+              .setChecksummedData(checksummedData.build());
+      if (!datum.isOnlyFullBlocks()) {
+        builder.setFinishWrite(true);
+        if (cumulative != null) {
+          builder.setObjectChecksums(
+              ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+        }
+        finished = true;
+      }
+
+      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
+      first = false;
+      messages.add(build);
+      bytesConsumed += contentSize;
+    }
+    if (finalize && !finished) {
+      messages.add(finishMessage(first));
+      finished = true;
+    }
+
+    try {
+      flush(messages);
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
+    }
+
+    return bytesConsumed;
+  }
+
+  @NonNull
+  private WriteObjectRequest finishMessage(boolean first) {
+    long offset = writeCtx.getTotalSentBytes().get();
+    Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+
+    WriteObjectRequest.Builder b =
+        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    if (crc32cValue != null) {
+      b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
+    }
+    WriteObjectRequest message = possiblyPairDownRequest(b, first).build();
+    return message;
+  }
+
+  private void flush(@NonNull List<WriteObjectRequest> segments) {
+    GrpcCallContext internalContext = contextWithBucketName(bucketName, baseContextSupplier.get());
+    ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> callable =
+        write.withDefaultCallContext(internalContext);
+
+    Retrying.run(
+        deps,
+        alg,
+        () -> {
+          Observer observer = new Observer(sizeCallback, completeCallback);
+          ApiStreamObserver<WriteObjectRequest> write = callable.clientStreamingCall(observer);
+
+          for (WriteObjectRequest message : segments) {
+            write.onNext(message);
+          }
+          write.onCompleted();
+          observer.await();
+          return null;
+        },
+        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> {
+
+    private final LongConsumer sizeCallback;
+    private final Consumer<WriteObjectResponse> completeCallback;
+
+    private final SettableApiFuture<Void> invocationHandle;
+    private volatile WriteObjectResponse last;
+
+    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
+      this.sizeCallback = sizeCallback;
+      this.completeCallback = completeCallback;
+      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);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        completeCallback.accept(last);
+      }
+      invocationHandle.set(null);
+    }
+
+    void await() {
+      try {
+        invocationHandle.get();
+      } catch (InterruptedException | ExecutionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        } else {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java
new file mode 100644
index 00000000..861e867d
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedDirectWritableByteChannel.java
@@ -0,0 +1,283 @@
+/*
+ * Copyright 2022 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.GrpcUtils.contextWithBucketName;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.cloud.storage.WriteCtx.SimpleWriteObjectRequestBuilderFactory;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.ObjectChecksums;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import java.util.function.LongConsumer;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class GapicUnbufferedDirectWritableByteChannel implements UnbufferedWritableByteChannel {
+
+  private final SettableApiFuture<WriteObjectResponse> resultFuture;
+  private final ChunkSegmenter chunkSegmenter;
+  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+
+  private final WriteCtx<SimpleWriteObjectRequestBuilderFactory> writeCtx;
+
+  private final Observer responseObserver;
+  private volatile ApiStreamObserver<WriteObjectRequest> stream;
+
+  private boolean open = true;
+  private boolean first = true;
+  private boolean finished = false;
+
+  GapicUnbufferedDirectWritableByteChannel(
+      SettableApiFuture<WriteObjectResponse> resultFuture,
+      ChunkSegmenter chunkSegmenter,
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      SimpleWriteObjectRequestBuilderFactory requestFactory) {
+    String bucketName = requestFactory.bucketName();
+    this.resultFuture = resultFuture;
+    this.chunkSegmenter = chunkSegmenter;
+
+    GrpcCallContext internalContext =
+        contextWithBucketName(bucketName, GrpcCallContext.createDefault());
+    this.write = write.withDefaultCallContext(internalContext);
+
+    this.writeCtx = new WriteCtx<>(requestFactory);
+    this.responseObserver = new Observer(writeCtx.getConfirmedBytes()::set, resultFuture::set);
+  }
+
+  @Override
+  public long write(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    return internalWrite(srcs, srcsOffset, srcsLength, false);
+  }
+
+  @Override
+  public long writeAndClose(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    long write = internalWrite(srcs, srcsOffset, srcsLength, true);
+    close();
+    return write;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
+    if (!finished) {
+      WriteObjectRequest message = finishMessage();
+      try {
+        openedStream.onNext(message);
+        openedStream.onCompleted();
+        finished = true;
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
+    } else {
+      try {
+        openedStream.onCompleted();
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
+    }
+    open = false;
+    responseObserver.await();
+  }
+
+  private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
+      throws ClosedChannelException {
+    if (!open) {
+      throw new ClosedChannelException();
+    }
+
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
+
+    List<WriteObjectRequest> messages = new ArrayList<>();
+
+    ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
+    int bytesConsumed = 0;
+    for (ChunkSegment datum : data) {
+      Crc32cLengthKnown crc32c = datum.getCrc32c();
+      ByteString b = datum.getB();
+      int contentSize = b.size();
+      long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+      Crc32cLengthKnown cumulative =
+          writeCtx
+              .getCumulativeCrc32c()
+              .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
+      ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
+      if (crc32c != null) {
+        checksummedData.setCrc32C(crc32c.getValue());
+      }
+      WriteObjectRequest.Builder builder =
+          writeCtx
+              .newRequestBuilder()
+              .setWriteOffset(offset)
+              .setChecksummedData(checksummedData.build());
+      if (!datum.isOnlyFullBlocks()) {
+        builder.setFinishWrite(true);
+        if (cumulative != null) {
+          builder.setObjectChecksums(
+              ObjectChecksums.newBuilder().setCrc32C(cumulative.getValue()).build());
+        }
+        finished = true;
+      }
+
+      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
+      first = false;
+      messages.add(build);
+      bytesConsumed += contentSize;
+    }
+    if (finalize && !finished) {
+      messages.add(finishMessage());
+      finished = true;
+    }
+
+    try {
+      for (WriteObjectRequest message : messages) {
+        openedStream.onNext(message);
+      }
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
+    }
+
+    return bytesConsumed;
+  }
+
+  @NonNull
+  private WriteObjectRequest finishMessage() {
+    long offset = writeCtx.getTotalSentBytes().get();
+    Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+
+    WriteObjectRequest.Builder b =
+        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    if (crc32cValue != null) {
+      b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
+    }
+    WriteObjectRequest message = possiblyPairDownRequest(b, first).build();
+    return message;
+  }
+
+  private ApiStreamObserver<WriteObjectRequest> openedStream() {
+    if (stream == null) {
+      synchronized (this) {
+        if (stream == null) {
+          stream = write.clientStreamingCall(responseObserver);
+        }
+      }
+    }
+    return stream;
+  }
+
+  /**
+   * Several fields of a WriteObjectRequest are only allowed on the "first" message sent to gcs,
+   * this utility method centralizes the logic necessary to clear those fields for use by subsequent
+   * messages.
+   */
+  private static WriteObjectRequest.Builder possiblyPairDownRequest(
+      WriteObjectRequest.Builder b, boolean firstMessageOfStream) {
+    if (firstMessageOfStream && b.getWriteOffset() == 0) {
+      return b;
+    }
+    if (b.getWriteOffset() > 0) {
+      b.clearWriteObjectSpec();
+    }
+
+    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
+      b.clearObjectChecksums();
+    }
+    return b;
+  }
+
+  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+
+    private final LongConsumer sizeCallback;
+    private final Consumer<WriteObjectResponse> completeCallback;
+
+    private final SettableApiFuture<Void> invocationHandle;
+    private volatile WriteObjectResponse last;
+
+    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
+      this.sizeCallback = sizeCallback;
+      this.completeCallback = completeCallback;
+      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);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        completeCallback.accept(last);
+      }
+      invocationHandle.set(null);
+    }
+
+    void await() {
+      try {
+        invocationHandle.get();
+      } catch (InterruptedException | ExecutionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        } else {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
similarity index 53%
rename from google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
rename to google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
index 32d5eb9d..16c6ec0a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel.java
@@ -16,14 +16,15 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
+
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
-import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
-import com.google.cloud.storage.WriteFlushStrategy.Flusher;
-import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -34,33 +35,42 @@ import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import java.util.function.LongConsumer;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
-final class GapicUnbufferedWritableByteChannel<
-        RequestFactoryT extends WriteObjectRequestBuilderFactory>
+final class GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel
     implements UnbufferedWritableByteChannel {
 
   private final SettableApiFuture<WriteObjectResponse> resultFuture;
   private final ChunkSegmenter chunkSegmenter;
+  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+
+  private final WriteCtx<ResumableWrite> writeCtx;
 
-  private final WriteCtx<RequestFactoryT> writeCtx;
-  private final Flusher flusher;
+  private final Observer responseObserver;
+  private volatile ApiStreamObserver<WriteObjectRequest> stream;
 
   private boolean open = true;
+  private boolean first = true;
   private boolean finished = false;
 
-  GapicUnbufferedWritableByteChannel(
+  GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
-      RequestFactoryT requestFactory,
-      FlusherFactory flusherFactory) {
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      ResumableWrite requestFactory) {
+    String bucketName = requestFactory.bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
+    GrpcCallContext internalContext =
+        contextWithBucketName(bucketName, GrpcCallContext.createDefault());
+    this.write = write.withDefaultCallContext(internalContext);
+
     this.writeCtx = new WriteCtx<>(requestFactory);
-    this.flusher =
-        flusherFactory.newFlusher(
-            requestFactory.bucketName(), writeCtx.getConfirmedBytes()::set, resultFuture::set);
+    this.responseObserver = new Observer(writeCtx.getConfirmedBytes()::set, resultFuture::set);
   }
 
   @Override
@@ -82,10 +92,12 @@ final class GapicUnbufferedWritableByteChannel<
 
   @Override
   public void close() throws IOException {
+    ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
     if (!finished) {
       WriteObjectRequest message = finishMessage();
       try {
-        flusher.close(message);
+        openedStream.onNext(message);
+        openedStream.onCompleted();
         finished = true;
       } catch (RuntimeException e) {
         resultFuture.setException(e);
@@ -93,18 +105,14 @@ final class GapicUnbufferedWritableByteChannel<
       }
     } else {
       try {
-        flusher.close(null);
+        openedStream.onCompleted();
       } catch (RuntimeException e) {
         resultFuture.setException(e);
         throw e;
       }
     }
     open = false;
-  }
-
-  @VisibleForTesting
-  WriteCtx<RequestFactoryT> getWriteCtx() {
-    return writeCtx;
+    responseObserver.await();
   }
 
   private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
@@ -117,6 +125,7 @@ final class GapicUnbufferedWritableByteChannel<
 
     List<WriteObjectRequest> messages = new ArrayList<>();
 
+    ApiStreamObserver<WriteObjectRequest> openedStream = openedStream();
     int bytesConsumed = 0;
     for (ChunkSegment datum : data) {
       Crc32cLengthKnown crc32c = datum.getCrc32c();
@@ -145,7 +154,8 @@ final class GapicUnbufferedWritableByteChannel<
         finished = true;
       }
 
-      WriteObjectRequest build = builder.build();
+      WriteObjectRequest build = possiblyPairDownRequest(builder, first).build();
+      first = false;
       messages.add(build);
       bytesConsumed += contentSize;
     }
@@ -155,7 +165,9 @@ final class GapicUnbufferedWritableByteChannel<
     }
 
     try {
-      flusher.flush(messages);
+      for (WriteObjectRequest message : messages) {
+        openedStream.onNext(message);
+      }
     } catch (RuntimeException e) {
       resultFuture.setException(e);
       throw e;
@@ -177,4 +189,95 @@ final class GapicUnbufferedWritableByteChannel<
     WriteObjectRequest message = b.build();
     return message;
   }
+
+  private ApiStreamObserver<WriteObjectRequest> openedStream() {
+    if (stream == null) {
+      synchronized (this) {
+        if (stream == null) {
+          stream = write.clientStreamingCall(responseObserver);
+        }
+      }
+    }
+    return stream;
+  }
+
+  /**
+   * Several fields of a WriteObjectRequest are only allowed on the "first" message sent to gcs,
+   * this utility method centralizes the logic necessary to clear those fields for use by subsequent
+   * messages.
+   */
+  private static WriteObjectRequest.Builder possiblyPairDownRequest(
+      WriteObjectRequest.Builder b, boolean firstMessageOfStream) {
+    if (firstMessageOfStream && b.getWriteOffset() == 0) {
+      return b;
+    }
+    if (b.getWriteOffset() > 0) {
+      b.clearWriteObjectSpec();
+    }
+
+    if (b.getWriteOffset() > 0 && !b.getFinishWrite()) {
+      b.clearObjectChecksums();
+    }
+    return b;
+  }
+
+  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+
+    private final LongConsumer sizeCallback;
+    private final Consumer<WriteObjectResponse> completeCallback;
+
+    private final SettableApiFuture<Void> invocationHandle;
+    private volatile WriteObjectResponse last;
+
+    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
+      this.sizeCallback = sizeCallback;
+      this.completeCallback = completeCallback;
+      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);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        completeCallback.accept(last);
+      }
+      invocationHandle.set(null);
+    }
+
+    void await() {
+      try {
+        invocationHandle.get();
+      } catch (InterruptedException | ExecutionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        } else {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
 }
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 5aa47eb6..1c6ad18c 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
@@ -30,7 +30,6 @@ import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
-import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
 import com.google.storage.v2.QueryWriteStatusRequest;
 import com.google.storage.v2.QueryWriteStatusResponse;
 import com.google.storage.v2.ServiceConstants.Values;
@@ -38,7 +37,7 @@ import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.nio.ByteBuffer;
 import java.util.function.BiFunction;
-import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 final class GapicWritableByteChannelSessionBuilder {
 
@@ -113,19 +112,7 @@ final class GapicWritableByteChannelSessionBuilder {
     return new JournalingResumableUploadBuilder();
   }
 
-  /**
-   * When constructing any of our channel sessions, there is always a {@link
-   * GapicUnbufferedWritableByteChannel} at the bottom of it. This method creates a BiFunction which
-   * will instantiate the {@link GapicUnbufferedWritableByteChannel} when provided with a {@code
-   * StartT} value and a {@code SettableApiFuture<WriteObjectResponse>}.
-   *
-   * <p>As part of providing the function, the provided parameters {@code FlusherFactory} and {@code
-   * f} are "bound" into the returned function. In conjunction with the configured fields of this
-   * class a new instance of {@link GapicUnbufferedWritableByteChannel} can be constructed.
-   */
-  private <StartT, RequestFactoryT extends WriteObjectRequestBuilderFactory>
-      BiFunction<StartT, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
-          bindFunction(FlusherFactory flusherFactory, Function<StartT, RequestFactoryT> f) {
+  private @NonNull ChunkSegmenter getChunkSegmenter() {
     // it is theoretically possible that the setter methods for the following variables could
     // be called again between when this method is invoked and the resulting function is invoked.
     // To ensure we are using the specified values at the point in time they are bound to the
@@ -133,12 +120,16 @@ final class GapicWritableByteChannelSessionBuilder {
     // fields.
     ByteStringStrategy boundStrategy = byteStringStrategy;
     Hasher boundHasher = hasher;
-    return (start, resultFuture) ->
-        new GapicUnbufferedWritableByteChannel<>(
-            resultFuture,
-            new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-            f.apply(start),
-            flusherFactory);
+    return new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE);
+  }
+
+  private static <StartT>
+      BiFunction<StartT, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
+          lift(
+              BiFunction<
+                      StartT, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
+                  func) {
+    return func;
   }
 
   final class DirectUploadBuilder {
@@ -189,9 +180,12 @@ final class GapicWritableByteChannelSessionBuilder {
       UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new UnbufferedWriteSession<>(
             ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-            bindFunction(
-                    WriteFlushStrategy.fsyncOnClose(write),
-                    WriteObjectRequestBuilderFactory::simple)
+            lift((WriteObjectRequest start, SettableApiFuture<WriteObjectResponse> resultFuture) ->
+                    new GapicUnbufferedDirectWritableByteChannel(
+                        resultFuture,
+                        getChunkSegmenter(),
+                        write,
+                        WriteObjectRequestBuilderFactory.simple(start)))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
     }
@@ -214,9 +208,12 @@ final class GapicWritableByteChannelSessionBuilder {
       BufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new BufferedWriteSession<>(
             ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-            bindFunction(
-                    WriteFlushStrategy.fsyncOnClose(write),
-                    WriteObjectRequestBuilderFactory::simple)
+            lift((WriteObjectRequest start, SettableApiFuture<WriteObjectResponse> resultFuture) ->
+                    new GapicUnbufferedDirectWritableByteChannel(
+                        resultFuture,
+                        getChunkSegmenter(),
+                        write,
+                        WriteObjectRequestBuilderFactory.simple(start)))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
@@ -296,12 +293,21 @@ final class GapicWritableByteChannelSessionBuilder {
       UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new UnbufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
-            bindFunction(
-                    fsyncEvery
-                        ? WriteFlushStrategy.fsyncEveryFlush(
-                            write, deps, alg, Retrying::newCallContext)
-                        : WriteFlushStrategy.fsyncOnClose(write),
-                    ResumableWrite::identity)
+            lift((ResumableWrite start, SettableApiFuture<WriteObjectResponse> result) -> {
+                  if (fsyncEvery) {
+                    return new GapicUnbufferedChunkedResumableWritableByteChannel(
+                        result,
+                        getChunkSegmenter(),
+                        write,
+                        ResumableWrite.identity(start),
+                        deps,
+                        alg,
+                        Retrying::newCallContext);
+                  } else {
+                    return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+                        result, getChunkSegmenter(), write, ResumableWrite.identity(start));
+                  }
+                })
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
     }
@@ -328,12 +334,21 @@ final class GapicWritableByteChannelSessionBuilder {
       BufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new BufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
-            bindFunction(
-                    fsyncEvery
-                        ? WriteFlushStrategy.fsyncEveryFlush(
-                            write, deps, alg, Retrying::newCallContext)
-                        : WriteFlushStrategy.fsyncOnClose(write),
-                    ResumableWrite::identity)
+            lift((ResumableWrite start, SettableApiFuture<WriteObjectResponse> result) -> {
+                  if (fsyncEvery) {
+                    return new GapicUnbufferedChunkedResumableWritableByteChannel(
+                        result,
+                        getChunkSegmenter(),
+                        write,
+                        ResumableWrite.identity(start),
+                        deps,
+                        alg,
+                        Retrying::newCallContext);
+                  } else {
+                    return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+                        result, getChunkSegmenter(), write, ResumableWrite.identity(start));
+                  }
+                })
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcUtils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcUtils.java
new file mode 100644
index 00000000..7e810115
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcUtils.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+final class GrpcUtils {
+
+  private GrpcUtils() {}
+
+  static GrpcCallContext contextWithBucketName(String bucketName, GrpcCallContext baseContext) {
+    if (bucketName != null && !bucketName.isEmpty()) {
+      return baseContext.withExtraHeaders(
+          ImmutableMap.of(
+              "x-goog-request-params", ImmutableList.of(String.format("bucket=%s", bucketName))));
+    }
+    return baseContext;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpWritableByteChannelSessionBuilder.java
index 19abf092..9626560e 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpWritableByteChannelSessionBuilder.java
@@ -113,17 +113,6 @@ final class HttpWritableByteChannelSessionBuilder {
       return new BufferedResumableUploadBuilder(bufferHandle);
     }
 
-    /**
-     * When constructing any of our channel sessions, there is always a {@link
-     * GapicUnbufferedWritableByteChannel} at the bottom of it. This method creates a BiFunction
-     * which will instantiate the {@link GapicUnbufferedWritableByteChannel} when provided with a
-     * {@code StartT} value and a {@code SettableApiFuture<WriteObjectResponse>}.
-     *
-     * <p>As part of providing the function, the provided parameters {@code FlusherFactory} and
-     * {@code f} are "bound" into the returned function. In conjunction with the configured fields
-     * of this class a new instance of {@link GapicUnbufferedWritableByteChannel} can be
-     * constructed.
-     */
     private BiFunction<
             JsonResumableWrite, SettableApiFuture<StorageObject>, UnbufferedWritableByteChannel>
         bindFunction() {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 0f6b002d..1a860a05 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -4825,7 +4825,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
    */
   @BetaApi
-  @TransportCompatibility({Transport.GRPC})
+  @TransportCompatibility({Transport.GRPC, Transport.HTTP})
   default BlobWriteSession blobWriteSession(BlobInfo blobInfo, BlobWriteOption... options) {
     return throwGrpcOnly(fmtMethodName("blobWriteSession", BlobInfo.class, BlobWriteOption.class));
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java
index 4169b841..67b5c089 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java
@@ -16,7 +16,7 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.WriteFlushStrategy.contextWithBucketName;
+import static com.google.cloud.storage.GrpcUtils.contextWithBucketName;
 
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
deleted file mode 100644
index 0a191c7d..00000000
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ /dev/null
@@ -1,521 +0,0 @@
-/*
- * Copyright 2022 Google LLC
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import com.google.api.core.SettableApiFuture;
-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.BidiStreamingCallable;
-import com.google.api.gax.rpc.ClientStreamingCallable;
-import com.google.cloud.storage.Conversions.Decoder;
-import com.google.cloud.storage.Retrying.RetryingDependencies;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.storage.v2.BidiWriteObjectRequest;
-import com.google.storage.v2.BidiWriteObjectResponse;
-import com.google.storage.v2.WriteObjectRequest;
-import com.google.storage.v2.WriteObjectResponse;
-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;
-
-/**
- * When writing to GCS using the WriteObject rpc, there are some behavioral differences between
- * performing a "direct" upload vs a "resumable" upload.
- *
- * <p>This class provides the encapsulation of the stream and "fsync" lifecycles and possible
- * automatic retry where applicable.
- *
- * <p>In this class "fsync" is used to mean "complete the client stream to GCS and await its
- * response". We are loosely following the concept used in linux to flush data to disk <a
- * target="_blank" href="https://man7.org/linux/man-pages/man2/fdatasync.2.html">fsync(2)</a>
- */
-final class WriteFlushStrategy {
-
-  private WriteFlushStrategy() {}
-
-  /**
-   * Create a {@link Flusher} which will "fsync" every time {@link Flusher#flush(List)} is called
-   * along with {@link Flusher#close(WriteObjectRequest)}.
-   */
-  static FlusherFactory fsyncEveryFlush(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      RetryingDependencies deps,
-      ResultRetryAlgorithm<?> alg,
-      Supplier<GrpcCallContext> baseContextSupplier) {
-    return (String bucketName,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncEveryFlusher(
-            write,
-            deps,
-            alg,
-            bucketName,
-            committedTotalBytesCallback,
-            onSuccessCallback,
-            baseContextSupplier);
-  }
-
-  /**
-   * Create a {@link BidiFlusher} which will keep a bidirectional stream open, flushing and sending
-   * the appropriate signals to GCS when the buffer is full.
-   */
-  static BidiFlusherFactory defaultBidiFlusher(
-      BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write,
-      RetryingDependencies deps,
-      ResultRetryAlgorithm<?> alg,
-      Supplier<GrpcCallContext> baseContextSupplier) {
-    return (String bucketName,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<BidiWriteObjectResponse> onSuccessCallback) ->
-        new DefaultBidiFlusher(
-            write,
-            deps,
-            alg,
-            bucketName,
-            committedTotalBytesCallback,
-            onSuccessCallback,
-            baseContextSupplier);
-  }
-
-  /**
-   * Create a {@link Flusher} which will "fsync" only on {@link Flusher#close(WriteObjectRequest)}.
-   * Calls to {@link Flusher#flush(List)} will be sent but not synced.
-   *
-   * @see FlusherFactory#newFlusher(String, LongConsumer, Consumer)
-   */
-  static FlusherFactory fsyncOnClose(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return (String bucketName,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncOnClose(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
-  }
-
-  static GrpcCallContext contextWithBucketName(String bucketName, GrpcCallContext baseContext) {
-    if (bucketName != null && !bucketName.isEmpty()) {
-      return baseContext.withExtraHeaders(
-          ImmutableMap.of(
-              "x-goog-request-params", ImmutableList.of(String.format("bucket=%s", bucketName))));
-    }
-    return baseContext;
-  }
-
-  /**
-   * 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 possiblyPairDownRequest(
-      WriteObjectRequest message, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && message.getWriteOffset() == 0) {
-      return message;
-    }
-
-    WriteObjectRequest.Builder b = message.toBuilder();
-    if (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (message.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (message.getWriteOffset() > 0 && !message.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b.build();
-  }
-
-  private static BidiWriteObjectRequest possiblyPairDownBidiRequest(
-      BidiWriteObjectRequest message, boolean firstMessageOfStream) {
-    if (firstMessageOfStream && message.getWriteOffset() == 0) {
-      return message;
-    }
-
-    BidiWriteObjectRequest.Builder b = message.toBuilder();
-    if (!firstMessageOfStream) {
-      b.clearUploadId();
-    }
-
-    if (message.getWriteOffset() > 0) {
-      b.clearWriteObjectSpec();
-    }
-
-    if (message.getWriteOffset() > 0 && !message.getFinishWrite()) {
-      b.clearObjectChecksums();
-    }
-    return b.build();
-  }
-
-  @FunctionalInterface
-  interface FlusherFactory {
-    /**
-     * @param committedTotalBytesCallback Callback to signal the total number of bytes committed by
-     *     this flusher.
-     * @param onSuccessCallback Callback to signal success, and provide the final response.
-     */
-    Flusher newFlusher(
-        String bucketName,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<WriteObjectResponse> onSuccessCallback);
-  }
-
-  interface Flusher {
-    void flush(@NonNull List<WriteObjectRequest> segments);
-
-    void close(@Nullable WriteObjectRequest req);
-  }
-
-  @FunctionalInterface
-  interface BidiFlusherFactory {
-    /**
-     * @param committedTotalBytesCallback Callback to signal the total number of bytes committed by
-     *     this flusher.
-     * @param onSuccessCallback Callback to signal success, and provide the final response.
-     */
-    BidiFlusher newFlusher(
-        String bucketName,
-        LongConsumer committedTotalBytesCallback,
-        Consumer<BidiWriteObjectResponse> onSuccessCallback);
-  }
-
-  interface BidiFlusher {
-    void flush(@NonNull List<BidiWriteObjectRequest> segments);
-
-    void close(@Nullable BidiWriteObjectRequest req);
-  }
-
-  private static final class FsyncEveryFlusher implements Flusher {
-
-    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
-    private final RetryingDependencies deps;
-    private final ResultRetryAlgorithm<?> alg;
-    private final String bucketName;
-    private final LongConsumer sizeCallback;
-    private final Consumer<WriteObjectResponse> completeCallback;
-    private final Supplier<GrpcCallContext> baseContextSupplier;
-
-    private FsyncEveryFlusher(
-        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-        RetryingDependencies deps,
-        ResultRetryAlgorithm<?> alg,
-        String bucketName,
-        LongConsumer sizeCallback,
-        Consumer<WriteObjectResponse> completeCallback,
-        Supplier<GrpcCallContext> baseContextSupplier) {
-      this.write = write;
-      this.deps = deps;
-      this.alg = alg;
-      this.bucketName = bucketName;
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
-      this.baseContextSupplier = baseContextSupplier;
-    }
-
-    public void flush(@NonNull List<WriteObjectRequest> segments) {
-      Retrying.run(
-          deps,
-          alg,
-          () -> {
-            Observer observer = new Observer(sizeCallback, completeCallback);
-            GrpcCallContext internalContext =
-                contextWithBucketName(bucketName, baseContextSupplier.get());
-            ApiStreamObserver<WriteObjectRequest> write =
-                this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
-
-            boolean first = true;
-            for (WriteObjectRequest message : segments) {
-              message = possiblyPairDownRequest(message, first);
-
-              write.onNext(message);
-              first = false;
-            }
-            write.onCompleted();
-            observer.await();
-            return null;
-          },
-          Decoder.identity());
-    }
-
-    public void close(@Nullable WriteObjectRequest req) {
-      if (req != null) {
-        flush(ImmutableList.of(req));
-      }
-    }
-  }
-
-  public static final class DefaultBidiFlusher implements BidiFlusher {
-
-    private final BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write;
-    private final RetryingDependencies deps;
-    private final ResultRetryAlgorithm<?> alg;
-    private final String bucketName;
-    private final LongConsumer sizeCallback;
-    private final Consumer<BidiWriteObjectResponse> completeCallback;
-    private final Supplier<GrpcCallContext> baseContextSupplier;
-    private volatile ApiStreamObserver<BidiWriteObjectRequest> stream;
-
-    private final BidiObserver responseObserver;
-
-    private DefaultBidiFlusher(
-        BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write,
-        RetryingDependencies deps,
-        ResultRetryAlgorithm<?> alg,
-        String bucketName,
-        LongConsumer sizeCallback,
-        Consumer<BidiWriteObjectResponse> completeCallback,
-        Supplier<GrpcCallContext> baseContextSupplier) {
-      this.write = write;
-      this.deps = deps;
-      this.alg = alg;
-      this.bucketName = bucketName;
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
-      this.baseContextSupplier = baseContextSupplier;
-      this.responseObserver = new BidiObserver(sizeCallback, completeCallback);
-    }
-
-    public void flush(@NonNull List<BidiWriteObjectRequest> segments) {
-      ensureOpen();
-      Retrying.run(
-          deps,
-          alg,
-          () -> {
-            boolean first = true;
-            for (BidiWriteObjectRequest message : segments) {
-              message = possiblyPairDownBidiRequest(message, first);
-
-              stream.onNext(message);
-              first = false;
-            }
-            BidiWriteObjectRequest message =
-                BidiWriteObjectRequest.newBuilder().setFlush(true).setStateLookup(true).build();
-            stream.onNext(message);
-            responseObserver.await();
-            return null;
-          },
-          Decoder.identity());
-    }
-
-    public void close(@Nullable BidiWriteObjectRequest req) {
-      ensureOpen();
-      if (req != null) {
-        flush(ImmutableList.of(req));
-      }
-    }
-
-    private void ensureOpen() {
-      if (stream == null) {
-        synchronized (this) {
-          if (stream == null) {
-            GrpcCallContext internalContext =
-                contextWithBucketName(bucketName, baseContextSupplier.get());
-            stream =
-                this.write
-                    .withDefaultCallContext(internalContext)
-                    .bidiStreamingCall(responseObserver);
-          }
-        }
-      }
-    }
-  }
-
-  private static final class FsyncOnClose implements Flusher {
-
-    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
-    private final String bucketName;
-    private final Observer responseObserver;
-
-    private volatile ApiStreamObserver<WriteObjectRequest> stream;
-    private boolean first = true;
-
-    private FsyncOnClose(
-        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-        String bucketName,
-        LongConsumer sizeCallback,
-        Consumer<WriteObjectResponse> completeCallback) {
-      this.write = write;
-      this.bucketName = bucketName;
-      this.responseObserver = new Observer(sizeCallback, completeCallback);
-    }
-
-    @Override
-    public void flush(@NonNull List<WriteObjectRequest> segments) {
-      ensureOpen();
-      for (WriteObjectRequest message : segments) {
-        message = possiblyPairDownRequest(message, first);
-
-        stream.onNext(message);
-        first = false;
-      }
-    }
-
-    @Override
-    public void close(@Nullable WriteObjectRequest message) {
-      ensureOpen();
-      if (message != null) {
-        message = possiblyPairDownRequest(message, first);
-        stream.onNext(message);
-      }
-      stream.onCompleted();
-      responseObserver.await();
-    }
-
-    private void ensureOpen() {
-      if (stream == null) {
-        synchronized (this) {
-          if (stream == null) {
-            GrpcCallContext internalContext =
-                contextWithBucketName(bucketName, GrpcCallContext.createDefault());
-            stream =
-                this.write
-                    .withDefaultCallContext(internalContext)
-                    .clientStreamingCall(responseObserver);
-          }
-        }
-      }
-    }
-  }
-
-  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
-
-    private final LongConsumer sizeCallback;
-    private final Consumer<WriteObjectResponse> completeCallback;
-
-    private final SettableApiFuture<Void> invocationHandle;
-    private volatile WriteObjectResponse last;
-
-    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
-      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);
-    }
-
-    @Override
-    public void onCompleted() {
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
-      }
-      invocationHandle.set(null);
-    }
-
-    void await() {
-      try {
-        invocationHandle.get();
-      } catch (InterruptedException | ExecutionException e) {
-        if (e.getCause() instanceof RuntimeException) {
-          throw (RuntimeException) e.getCause();
-        } else {
-          throw new RuntimeException(e);
-        }
-      }
-    }
-  }
-
-  static class BidiObserver implements ApiStreamObserver<BidiWriteObjectResponse> {
-
-    private final LongConsumer sizeCallback;
-    private final Consumer<BidiWriteObjectResponse> completeCallback;
-
-    private final SettableApiFuture<Void> invocationHandle;
-    private volatile BidiWriteObjectResponse last;
-
-    BidiObserver(LongConsumer sizeCallback, Consumer<BidiWriteObjectResponse> completeCallback) {
-      this.sizeCallback = sizeCallback;
-      this.completeCallback = completeCallback;
-      this.invocationHandle = SettableApiFuture.create();
-    }
-
-    @Override
-    public void onNext(BidiWriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        sizeCallback.accept(value.getPersistedSize());
-        invocationHandle.set(null);
-      } 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);
-    }
-
-    @Override
-    public void onCompleted() {
-
-      if (last != null && last.hasResource()) {
-        completeCallback.accept(last);
-      }
-      invocationHandle.set(null);
-    }
-
-    void await() {
-      try {
-        invocationHandle.get();
-      } catch (InterruptedException | ExecutionException e) {
-        if (e.getCause() instanceof RuntimeException) {
-          throw (RuntimeException) e.getCause();
-        } else {
-          throw new RuntimeException(e);
-        }
-      }
-    }
-  }
-}
diff --git a/grpc-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/StorageControlGrpc.java b/grpc-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/StorageControlGrpc.java
index 476945a1..9456b817 100644
--- a/grpc-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/StorageControlGrpc.java
+++ b/grpc-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/StorageControlGrpc.java
@@ -539,7 +539,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Creates a new folder.
+     * Creates a new folder. This operation is only applicable to a hierarchical
+     * namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     default void createFolder(
@@ -553,7 +555,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Permanently deletes an empty folder.
+     * Permanently deletes an empty folder. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     default void deleteFolder(
@@ -567,7 +571,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Returns metadata for the specified folder.
+     * Returns metadata for the specified folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     default void getFolder(
@@ -580,7 +586,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Retrieves a list of folders for a given bucket.
+     * Retrieves a list of folders. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     default void listFolders(
@@ -595,9 +603,11 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Renames a source folder to a destination folder. During a rename, the
+     * Renames a source folder to a destination folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket. During a rename, the
      * source and destination folders are locked until the long running operation
      * completes.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     default void renameFolder(
@@ -717,7 +727,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Creates a new folder.
+     * Creates a new folder. This operation is only applicable to a hierarchical
+     * namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public void createFolder(
@@ -733,7 +745,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Permanently deletes an empty folder.
+     * Permanently deletes an empty folder. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public void deleteFolder(
@@ -749,7 +763,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Returns metadata for the specified folder.
+     * Returns metadata for the specified folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public void getFolder(
@@ -763,7 +779,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Retrieves a list of folders for a given bucket.
+     * Retrieves a list of folders. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public void listFolders(
@@ -780,9 +798,11 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Renames a source folder to a destination folder. During a rename, the
+     * Renames a source folder to a destination folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket. During a rename, the
      * source and destination folders are locked until the long running operation
      * completes.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public void renameFolder(
@@ -899,7 +919,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Creates a new folder.
+     * Creates a new folder. This operation is only applicable to a hierarchical
+     * namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.storage.control.v2.Folder createFolder(
@@ -912,7 +934,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Permanently deletes an empty folder.
+     * Permanently deletes an empty folder. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.protobuf.Empty deleteFolder(
@@ -925,7 +949,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Returns metadata for the specified folder.
+     * Returns metadata for the specified folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.storage.control.v2.Folder getFolder(
@@ -938,7 +964,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Retrieves a list of folders for a given bucket.
+     * Retrieves a list of folders. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.storage.control.v2.ListFoldersResponse listFolders(
@@ -951,9 +979,11 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Renames a source folder to a destination folder. During a rename, the
+     * Renames a source folder to a destination folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket. During a rename, the
      * source and destination folders are locked until the long running operation
      * completes.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.longrunning.Operation renameFolder(
@@ -1051,7 +1081,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Creates a new folder.
+     * Creates a new folder. This operation is only applicable to a hierarchical
+     * namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.common.util.concurrent.ListenableFuture<com.google.storage.control.v2.Folder>
@@ -1064,7 +1096,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Permanently deletes an empty folder.
+     * Permanently deletes an empty folder. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.common.util.concurrent.ListenableFuture<com.google.protobuf.Empty>
@@ -1077,7 +1111,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Returns metadata for the specified folder.
+     * Returns metadata for the specified folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.common.util.concurrent.ListenableFuture<com.google.storage.control.v2.Folder>
@@ -1090,7 +1126,9 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Retrieves a list of folders for a given bucket.
+     * Retrieves a list of folders. This operation is only applicable to a
+     * hierarchical namespace enabled bucket.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.common.util.concurrent.ListenableFuture<
@@ -1104,9 +1142,11 @@ public final class StorageControlGrpc {
      *
      *
      * <pre>
-     * Renames a source folder to a destination folder. During a rename, the
+     * Renames a source folder to a destination folder. This operation is only
+     * applicable to a hierarchical namespace enabled bucket. During a rename, the
      * source and destination folders are locked until the long running operation
      * completes.
+     * Hierarchical namespace buckets are in allowlist preview.
      * </pre>
      */
     public com.google.common.util.concurrent.ListenableFuture<com.google.longrunning.Operation>
diff --git a/pom.xml b/pom.xml
index 9c4a9c87..d7c02eff 100644
--- a/pom.xml
+++ b/pom.xml
@@ -86,7 +86,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-pubsub</artifactId>
-        <version>1.129.0</version>
+        <version>1.129.2</version>
         <scope>test</scope>
       </dependency>
       <dependency>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequest.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequest.java
index 9fe4ed2e..0686e2ed 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequest.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequest.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * Request message for CreateFolder.
+ * Request message for CreateFolder. This operation is only applicable to a
+ * hierarchical namespace enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.CreateFolderRequest}
@@ -74,7 +76,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
    *
    *
    * <pre>
-   * Required. Name of the bucket in which the folder will reside.
+   * Required. Name of the bucket in which the folder will reside. The bucket
+   * must be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -99,7 +102,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
    *
    *
    * <pre>
-   * Required. Name of the bucket in which the folder will reside.
+   * Required. Name of the bucket in which the folder will reside. The bucket
+   * must be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -266,8 +270,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -293,8 +296,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -520,7 +522,9 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
    *
    *
    * <pre>
-   * Request message for CreateFolder.
+   * Request message for CreateFolder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.CreateFolderRequest}
@@ -776,7 +780,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      *
      * <pre>
-     * Required. Name of the bucket in which the folder will reside.
+     * Required. Name of the bucket in which the folder will reside. The bucket
+     * must be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -800,7 +805,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      *
      * <pre>
-     * Required. Name of the bucket in which the folder will reside.
+     * Required. Name of the bucket in which the folder will reside. The bucket
+     * must be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -824,7 +830,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      *
      * <pre>
-     * Required. Name of the bucket in which the folder will reside.
+     * Required. Name of the bucket in which the folder will reside. The bucket
+     * must be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -847,7 +854,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      *
      * <pre>
-     * Required. Name of the bucket in which the folder will reside.
+     * Required. Name of the bucket in which the folder will reside. The bucket
+     * must be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -866,7 +874,8 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      *
      * <pre>
-     * Required. Name of the bucket in which the folder will reside.
+     * Required. Name of the bucket in which the folder will reside. The bucket
+     * must be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -1296,8 +1305,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1322,8 +1330,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1348,8 +1355,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1373,8 +1379,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1394,8 +1399,7 @@ public final class CreateFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequestOrBuilder.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequestOrBuilder.java
index 38b0041b..88c7f84c 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequestOrBuilder.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/CreateFolderRequestOrBuilder.java
@@ -28,7 +28,8 @@ public interface CreateFolderRequestOrBuilder
    *
    *
    * <pre>
-   * Required. Name of the bucket in which the folder will reside.
+   * Required. Name of the bucket in which the folder will reside. The bucket
+   * must be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -42,7 +43,8 @@ public interface CreateFolderRequestOrBuilder
    *
    *
    * <pre>
-   * Required. Name of the bucket in which the folder will reside.
+   * Required. Name of the bucket in which the folder will reside. The bucket
+   * must be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -152,8 +154,7 @@ public interface CreateFolderRequestOrBuilder
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -168,8 +169,7 @@ public interface CreateFolderRequestOrBuilder
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequest.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequest.java
index 5c702102..8c53dc04 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequest.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequest.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * Request message for DeleteFolder.
+ * Request message for DeleteFolder. This operation is only applicable to a
+ * hierarchical namespace enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.DeleteFolderRequest}
@@ -201,8 +203,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -228,8 +229,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -451,7 +451,9 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
    *
    *
    * <pre>
-   * Request message for DeleteFolder.
+   * Request message for DeleteFolder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.DeleteFolderRequest}
@@ -946,8 +948,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -972,8 +973,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -998,8 +998,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1023,8 +1022,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
@@ -1044,8 +1042,7 @@ public final class DeleteFolderRequest extends com.google.protobuf.GeneratedMess
      *
      * <pre>
      * Optional. A unique identifier for this request. UUID is the recommended
-     * format, but other formats are still accepted. This request is only
-     * idempotent if a `request_id` is provided.
+     * format, but other formats are still accepted.
      * </pre>
      *
      * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequestOrBuilder.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequestOrBuilder.java
index 2e7004fd..3f7b5b62 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequestOrBuilder.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/DeleteFolderRequestOrBuilder.java
@@ -114,8 +114,7 @@ public interface DeleteFolderRequestOrBuilder
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
@@ -130,8 +129,7 @@ public interface DeleteFolderRequestOrBuilder
    *
    * <pre>
    * Optional. A unique identifier for this request. UUID is the recommended
-   * format, but other formats are still accepted. This request is only
-   * idempotent if a `request_id` is provided.
+   * format, but other formats are still accepted.
    * </pre>
    *
    * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/Folder.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/Folder.java
index c05fc457..5a4ec2f5 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/Folder.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/Folder.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * A folder.
+ * A folder resource. This resource can only exist in a hierarchical namespace
+ * enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.Folder}
@@ -514,7 +516,9 @@ public final class Folder extends com.google.protobuf.GeneratedMessageV3
    *
    *
    * <pre>
-   * A folder.
+   * A folder resource. This resource can only exist in a hierarchical namespace
+   * enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.Folder}
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/GetFolderRequest.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/GetFolderRequest.java
index abaa8c2a..6baf949a 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/GetFolderRequest.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/GetFolderRequest.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * Request message for GetFolder.
+ * Request message for GetFolder. This operation is only applicable to a
+ * hierarchical namespace enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.GetFolderRequest}
@@ -449,7 +451,9 @@ public final class GetFolderRequest extends com.google.protobuf.GeneratedMessage
    *
    *
    * <pre>
-   * Request message for GetFolder.
+   * Request message for GetFolder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.GetFolderRequest}
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequest.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequest.java
index 0d8481b8..e47352ba 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequest.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequest.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * Request message for ListFolders.
+ * Request message for ListFolders. This operation is only applicable to a
+ * hierarchical namespace enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.ListFoldersRequest}
@@ -77,7 +79,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
    *
    *
    * <pre>
-   * Required. Name of the bucket in which to look for folders.
+   * Required. Name of the bucket in which to look for folders. The bucket must
+   * be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -102,7 +105,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
    *
    *
    * <pre>
-   * Required. Name of the bucket in which to look for folders.
+   * Required. Name of the bucket in which to look for folders. The bucket must
+   * be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -701,7 +705,9 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
    *
    *
    * <pre>
-   * Request message for ListFolders.
+   * Request message for ListFolders. This operation is only applicable to a
+   * hierarchical namespace enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.ListFoldersRequest}
@@ -988,7 +994,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
      *
      *
      * <pre>
-     * Required. Name of the bucket in which to look for folders.
+     * Required. Name of the bucket in which to look for folders. The bucket must
+     * be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -1012,7 +1019,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
      *
      *
      * <pre>
-     * Required. Name of the bucket in which to look for folders.
+     * Required. Name of the bucket in which to look for folders. The bucket must
+     * be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -1036,7 +1044,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
      *
      *
      * <pre>
-     * Required. Name of the bucket in which to look for folders.
+     * Required. Name of the bucket in which to look for folders. The bucket must
+     * be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -1059,7 +1068,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
      *
      *
      * <pre>
-     * Required. Name of the bucket in which to look for folders.
+     * Required. Name of the bucket in which to look for folders. The bucket must
+     * be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
@@ -1078,7 +1088,8 @@ public final class ListFoldersRequest extends com.google.protobuf.GeneratedMessa
      *
      *
      * <pre>
-     * Required. Name of the bucket in which to look for folders.
+     * Required. Name of the bucket in which to look for folders. The bucket must
+     * be a hierarchical namespace enabled bucket.
      * </pre>
      *
      * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequestOrBuilder.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequestOrBuilder.java
index 113d2ff3..0e06b432 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequestOrBuilder.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/ListFoldersRequestOrBuilder.java
@@ -28,7 +28,8 @@ public interface ListFoldersRequestOrBuilder
    *
    *
    * <pre>
-   * Required. Name of the bucket in which to look for folders.
+   * Required. Name of the bucket in which to look for folders. The bucket must
+   * be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
@@ -42,7 +43,8 @@ public interface ListFoldersRequestOrBuilder
    *
    *
    * <pre>
-   * Required. Name of the bucket in which to look for folders.
+   * Required. Name of the bucket in which to look for folders. The bucket must
+   * be a hierarchical namespace enabled bucket.
    * </pre>
    *
    * <code>
diff --git a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/RenameFolderRequest.java b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/RenameFolderRequest.java
index 46a2ba44..aed982fd 100644
--- a/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/RenameFolderRequest.java
+++ b/proto-google-cloud-storage-control-v2/src/main/java/com/google/storage/control/v2/RenameFolderRequest.java
@@ -23,7 +23,9 @@ package com.google.storage.control.v2;
  *
  *
  * <pre>
- * Request message for RenameFolder.
+ * Request message for RenameFolder. This operation is only applicable to a
+ * hierarchical namespace enabled bucket.
+ * Hierarchical namespace buckets are in allowlist preview.
  * </pre>
  *
  * Protobuf type {@code google.storage.control.v2.RenameFolderRequest}
@@ -512,7 +514,9 @@ public final class RenameFolderRequest extends com.google.protobuf.GeneratedMess
    *
    *
    * <pre>
-   * Request message for RenameFolder.
+   * Request message for RenameFolder. This operation is only applicable to a
+   * hierarchical namespace enabled bucket.
+   * Hierarchical namespace buckets are in allowlist preview.
    * </pre>
    *
    * Protobuf type {@code google.storage.control.v2.RenameFolderRequest}
diff --git a/proto-google-cloud-storage-control-v2/src/main/proto/google/storage/control/v2/storage_control.proto b/proto-google-cloud-storage-control-v2/src/main/proto/google/storage/control/v2/storage_control.proto
index 5e460eb9..61663a64 100644
--- a/proto-google-cloud-storage-control-v2/src/main/proto/google/storage/control/v2/storage_control.proto
+++ b/proto-google-cloud-storage-control-v2/src/main/proto/google/storage/control/v2/storage_control.proto
@@ -47,7 +47,9 @@ service StorageControl {
       "https://www.googleapis.com/auth/devstorage.read_only,"
       "https://www.googleapis.com/auth/devstorage.read_write";
 
-  // Creates a new folder.
+  // Creates a new folder. This operation is only applicable to a hierarchical
+  // namespace enabled bucket.
+  // Hierarchical namespace buckets are in allowlist preview.
   rpc CreateFolder(CreateFolderRequest) returns (Folder) {
     option (google.api.routing) = {
       routing_parameters { field: "parent" path_template: "{bucket=**}" }
@@ -55,7 +57,9 @@ service StorageControl {
     option (google.api.method_signature) = "parent,folder,folder_id";
   }
 
-  // Permanently deletes an empty folder.
+  // Permanently deletes an empty folder. This operation is only applicable to a
+  // hierarchical namespace enabled bucket.
+  // Hierarchical namespace buckets are in allowlist preview.
   rpc DeleteFolder(DeleteFolderRequest) returns (google.protobuf.Empty) {
     option (google.api.routing) = {
       routing_parameters {
@@ -66,7 +70,9 @@ service StorageControl {
     option (google.api.method_signature) = "name";
   }
 
-  // Returns metadata for the specified folder.
+  // Returns metadata for the specified folder. This operation is only
+  // applicable to a hierarchical namespace enabled bucket.
+  // Hierarchical namespace buckets are in allowlist preview.
   rpc GetFolder(GetFolderRequest) returns (Folder) {
     option (google.api.routing) = {
       routing_parameters {
@@ -77,7 +83,9 @@ service StorageControl {
     option (google.api.method_signature) = "name";
   }
 
-  // Retrieves a list of folders for a given bucket.
+  // Retrieves a list of folders. This operation is only applicable to a
+  // hierarchical namespace enabled bucket.
+  // Hierarchical namespace buckets are in allowlist preview.
   rpc ListFolders(ListFoldersRequest) returns (ListFoldersResponse) {
     option (google.api.routing) = {
       routing_parameters { field: "parent" path_template: "{bucket=**}" }
@@ -85,9 +93,11 @@ service StorageControl {
     option (google.api.method_signature) = "parent";
   }
 
-  // Renames a source folder to a destination folder. During a rename, the
+  // Renames a source folder to a destination folder. This operation is only
+  // applicable to a hierarchical namespace enabled bucket. During a rename, the
   // source and destination folders are locked until the long running operation
   // completes.
+  // Hierarchical namespace buckets are in allowlist preview.
   rpc RenameFolder(RenameFolderRequest) returns (google.longrunning.Operation) {
     option (google.api.routing) = {
       routing_parameters {
@@ -161,7 +171,9 @@ message PendingRenameInfo {
   string operation = 1 [(google.api.field_behavior) = OUTPUT_ONLY];
 }
 
-// A folder.
+// A folder resource. This resource can only exist in a hierarchical namespace
+// enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message Folder {
   option (google.api.resource) = {
     type: "storage.googleapis.com/Folder"
@@ -195,7 +207,9 @@ message Folder {
       [(google.api.field_behavior) = OUTPUT_ONLY];
 }
 
-// Request message for GetFolder.
+// Request message for GetFolder. This operation is only applicable to a
+// hierarchical namespace enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message GetFolderRequest {
   // Required. Name of the folder.
   // Format: `projects/{project}/buckets/{bucket}/folders/{folder}`
@@ -220,9 +234,12 @@ message GetFolderRequest {
   ];
 }
 
-// Request message for CreateFolder.
+// Request message for CreateFolder. This operation is only applicable to a
+// hierarchical namespace enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message CreateFolderRequest {
-  // Required. Name of the bucket in which the folder will reside.
+  // Required. Name of the bucket in which the folder will reside. The bucket
+  // must be a hierarchical namespace enabled bucket.
   string parent = 1 [
     (google.api.field_behavior) = REQUIRED,
     (google.api.resource_reference) = {
@@ -248,15 +265,16 @@ message CreateFolderRequest {
   bool recursive = 4 [(google.api.field_behavior) = OPTIONAL];
 
   // Optional. A unique identifier for this request. UUID is the recommended
-  // format, but other formats are still accepted. This request is only
-  // idempotent if a `request_id` is provided.
+  // format, but other formats are still accepted.
   string request_id = 5 [
     (google.api.field_info).format = UUID4,
     (google.api.field_behavior) = OPTIONAL
   ];
 }
 
-// Request message for DeleteFolder.
+// Request message for DeleteFolder. This operation is only applicable to a
+// hierarchical namespace enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message DeleteFolderRequest {
   // Required. Name of the folder.
   // Format: `projects/{project}/buckets/{bucket}/folders/{folder}`
@@ -274,17 +292,19 @@ message DeleteFolderRequest {
   optional int64 if_metageneration_not_match = 4;
 
   // Optional. A unique identifier for this request. UUID is the recommended
-  // format, but other formats are still accepted. This request is only
-  // idempotent if a `request_id` is provided.
+  // format, but other formats are still accepted.
   string request_id = 5 [
     (google.api.field_info).format = UUID4,
     (google.api.field_behavior) = OPTIONAL
   ];
 }
 
-// Request message for ListFolders.
+// Request message for ListFolders. This operation is only applicable to a
+// hierarchical namespace enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message ListFoldersRequest {
-  // Required. Name of the bucket in which to look for folders.
+  // Required. Name of the bucket in which to look for folders. The bucket must
+  // be a hierarchical namespace enabled bucket.
   string parent = 1 [
     (google.api.field_behavior) = REQUIRED,
     (google.api.resource_reference) = {
@@ -339,7 +359,9 @@ message ListFoldersResponse {
   string next_page_token = 2;
 }
 
-// Request message for RenameFolder.
+// Request message for RenameFolder. This operation is only applicable to a
+// hierarchical namespace enabled bucket.
+// Hierarchical namespace buckets are in allowlist preview.
 message RenameFolderRequest {
   // Required. Name of the source folder being renamed.
   // Format: `projects/{project}/buckets/{bucket}/folders/{folder}`
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 3b3024f9..610ecce6 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -30,7 +30,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.36.1</version>
+      <version>2.37.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
@@ -66,13 +66,13 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.0</version>
+      <version>1.129.2</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage-control</artifactId>
-      <version>2.36.1-alpha</version>
+      <version>2.37.0-alpha</version>
       <scope>compile</scope>
     </dependency>
   </dependencies>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index 967c12f6..f02de229 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -29,7 +29,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.37.0</version>
+        <version>26.38.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -61,7 +61,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.0</version>
+      <version>1.129.2</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -78,7 +78,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-jar-plugin</artifactId>
-        <version>3.4.0</version>
+        <version>3.4.1</version>
         <configuration>
           <archive>
             <manifest>
diff --git a/samples/pom.xml b/samples/pom.xml
index 0490423a..c1b0d76c 100644
--- a/samples/pom.xml
+++ b/samples/pom.xml
@@ -40,7 +40,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-deploy-plugin</artifactId>
-        <version>3.1.1</version>
+        <version>3.1.2</version>
         <configuration>
           <skip>true</skip>
         </configuration>
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index cf503809..b85c897a 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -58,7 +58,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.0</version>
+      <version>1.129.2</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index e7983001..f87f8d33 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.37.0</version>
+        <version>26.38.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -76,7 +76,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-pubsub</artifactId>
-      <version>1.129.0</version>
+      <version>1.129.2</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowDivideAndConquerDownload.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowDivideAndConquerDownload.java
new file mode 100644
index 00000000..fc1921a2
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowDivideAndConquerDownload.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package com.example.storage.transfermanager;
+
+// [START storage_transfer_manager_download_chunks_concurrently]
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.transfermanager.DownloadResult;
+import com.google.cloud.storage.transfermanager.ParallelDownloadConfig;
+import com.google.cloud.storage.transfermanager.TransferManager;
+import com.google.cloud.storage.transfermanager.TransferManagerConfig;
+import java.nio.file.Path;
+import java.util.List;
+
+class AllowDivideAndConquerDownload {
+
+  public static void divideAndConquerDownloadAllowed(List<BlobInfo> blobs,
+      String bucketName, Path destinationDirectory) {
+    TransferManager transferManager = TransferManagerConfig.newBuilder()
+        .setAllowDivideAndConquerDownload(true)
+        .build()
+        .getService();
+    ParallelDownloadConfig parallelDownloadConfig = ParallelDownloadConfig.newBuilder()
+        .setBucketName(bucketName)
+        .setDownloadDirectory(destinationDirectory)
+        .build();
+    List<DownloadResult> results = transferManager
+        .downloadBlobs(blobs, parallelDownloadConfig)
+        .getDownloadResults();
+
+    for (DownloadResult result : results) {
+      System.out.println("Download of " + result.getInput().getName()
+          + " completed with status "
+          + result.getStatus());
+    }
+
+  }
+}
+// [END storage_transfer_manager_download_chunks_concurrently]
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowParallelCompositeUpload.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowParallelCompositeUpload.java
new file mode 100644
index 00000000..e1ac5134
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/AllowParallelCompositeUpload.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package com.example.storage.transfermanager;
+
+// [START storage_transfer_manager_upload_chunks_concurrently]
+import com.google.cloud.storage.transfermanager.ParallelUploadConfig;
+import com.google.cloud.storage.transfermanager.TransferManager;
+import com.google.cloud.storage.transfermanager.TransferManagerConfig;
+import com.google.cloud.storage.transfermanager.UploadResult;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+
+class AllowParallelCompositeUpload {
+
+  public static void parallelCompositeUploadAllowed(String bucketName, List<Path> files)
+      throws IOException {
+    TransferManager transferManager = TransferManagerConfig
+        .newBuilder()
+        .setAllowParallelCompositeUpload(true)
+        .build()
+        .getService();
+    ParallelUploadConfig parallelUploadConfig =
+        ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
+    List<UploadResult> results =
+        transferManager.uploadFiles(files, parallelUploadConfig).getUploadResults();
+    for (UploadResult result : results) {
+      System.out.println(
+          "Upload for "
+              + result.getInput().getName()
+              + " completed with status "
+              + result.getStatus());
+    }
+  }
+}
+// [END storage_transfer_manager_upload_chunks_concurrently]


Test patch:
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 3054616c..edbc8006 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
@@ -25,8 +25,8 @@ import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.rpc.DataLossException;
 import com.google.api.gax.rpc.PermissionDeniedException;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.WriteCtx.SimpleWriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
-import com.google.cloud.storage.WriteFlushStrategy.Flusher;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -51,13 +51,10 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.logging.Logger;
 import java.util.stream.Collector;
 import java.util.stream.Collectors;
-import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Test;
 
 public final class ITGapicUnbufferedWritableByteChannelTest {
@@ -124,7 +121,7 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
   private static final WriteObjectResponse resp5 =
       WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
 
-  private static final WriteObjectRequestBuilderFactory reqFactory =
+  private static final ResumableWrite reqFactory =
       new ResumableWrite(startReq, startResp, TestUtils.onlyUploadId());
 
   @Test
@@ -149,7 +146,8 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
     WriteObjectResponse resp = resp5;
 
     WriteObjectRequest base = WriteObjectRequest.newBuilder().setWriteObjectSpec(spec).build();
-    WriteObjectRequestBuilderFactory reqFactory = WriteObjectRequestBuilderFactory.simple(base);
+    SimpleWriteObjectRequestBuilderFactory reqFactory =
+        WriteObjectRequestBuilderFactory.simple(base);
 
     StorageImplBase service =
         new DirectWriteService(
@@ -157,12 +155,9 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
     try (FakeServer fake = FakeServer.of(service);
         StorageClient sc = StorageClient.create(fake.storageSettings())) {
       SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
-      try (GapicUnbufferedWritableByteChannel<?> c =
-          new GapicUnbufferedWritableByteChannel<>(
-              result,
-              segmenter,
-              reqFactory,
-              WriteFlushStrategy.fsyncOnClose(sc.writeObjectCallable()))) {
+      try (GapicUnbufferedDirectWritableByteChannel c =
+          new GapicUnbufferedDirectWritableByteChannel(
+              result, segmenter, sc.writeObjectCallable(), reqFactory)) {
         c.write(ByteBuffer.wrap(bytes));
       }
       assertThat(result.get()).isEqualTo(resp);
@@ -183,16 +178,15 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
     try (FakeServer fake = FakeServer.of(service);
         StorageClient sc = StorageClient.create(fake.storageSettings())) {
       SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
-      GapicUnbufferedWritableByteChannel<?> c =
-          new GapicUnbufferedWritableByteChannel<>(
+      GapicUnbufferedChunkedResumableWritableByteChannel c =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
               result,
               segmenter,
+              sc.writeObjectCallable(),
               reqFactory,
-              WriteFlushStrategy.fsyncEveryFlush(
-                  sc.writeObjectCallable(),
-                  RetryingDependencies.attemptOnce(),
-                  Retrying.neverRetry(),
-                  Retrying::newCallContext));
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              Retrying::newCallContext);
       ArrayList<String> debugMessages = new ArrayList<>();
       try {
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
@@ -267,21 +261,20 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
     try (FakeServer fake = FakeServer.of(service);
         StorageClient sc = StorageClient.create(fake.storageSettings())) {
       SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
-      try (GapicUnbufferedWritableByteChannel<?> c =
-          new GapicUnbufferedWritableByteChannel<>(
+      try (GapicUnbufferedChunkedResumableWritableByteChannel c =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
               result,
               segmenter,
+              sc.writeObjectCallable(),
               reqFactory,
-              WriteFlushStrategy.fsyncEveryFlush(
-                  sc.writeObjectCallable(),
-                  TestUtils.defaultRetryingDeps(),
-                  new BasicResultRetryAlgorithm<Object>() {
-                    @Override
-                    public boolean shouldRetry(Throwable t, Object ignore) {
-                      return TestUtils.findThrowable(DataLossException.class, t) != null;
-                    }
-                  },
-                  Retrying::newCallContext))) {
+              TestUtils.defaultRetryingDeps(),
+              new BasicResultRetryAlgorithm<Object>() {
+                @Override
+                public boolean shouldRetry(Throwable t, Object ignore) {
+                  return TestUtils.findThrowable(DataLossException.class, t) != null;
+                }
+              },
+              Retrying::newCallContext)) {
         writeCtx = c.getWriteCtx();
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
         c.write(buffers.get(0));
@@ -309,37 +302,38 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
 
   @Test
   public void resumableUpload_finalizeWhenWriteAndCloseCalledEvenWhenQuantumAligned()
-      throws IOException {
-    SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
-
-    AtomicReference<List<WriteObjectRequest>> actualFlush = new AtomicReference<>();
-    WriteObjectRequest closeRequestSentinel =
-        WriteObjectRequest.newBuilder().setUploadId("sentinel").build();
-    AtomicReference<WriteObjectRequest> actualClose = new AtomicReference<>(closeRequestSentinel);
-    GapicUnbufferedWritableByteChannel<?> c =
-        new GapicUnbufferedWritableByteChannel<>(
-            result,
-            segmenter,
-            reqFactory,
-            (bucketName, committedTotalBytesCallback, onSuccessCallback) ->
-                new Flusher() {
-                  @Override
-                  public void flush(@NonNull List<WriteObjectRequest> segments) {
-                    actualFlush.compareAndSet(null, segments);
-                  }
-
-                  @Override
-                  public void close(@Nullable WriteObjectRequest req) {
-                    actualClose.compareAndSet(closeRequestSentinel, req);
-                  }
-                });
-
-    long written = c.writeAndClose(ByteBuffer.wrap(bytes));
-
-    assertThat(written).isEqualTo(40);
-    assertThat(actualFlush.get()).isEqualTo(ImmutableList.of(req1, req2, req3, req4, req5));
-    // calling close is okay, as long as the provided request is null
-    assertThat(actualClose.get()).isAnyOf(closeRequestSentinel, null);
+      throws IOException, InterruptedException, ExecutionException {
+    ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes =
+        ImmutableMap.<List<WriteObjectRequest>, WriteObjectResponse>builder()
+            .put(
+                ImmutableList.of(
+                    req1,
+                    req2.toBuilder().clearUploadId().build(),
+                    req3.toBuilder().clearUploadId().build(),
+                    req4.toBuilder().clearUploadId().build(),
+                    req5.toBuilder().clearUploadId().build()),
+                resp5)
+            .build();
+    StorageImplBase service = new DirectWriteService(writes);
+    try (FakeServer fake = FakeServer.of(service);
+        StorageClient sc = StorageClient.create(fake.storageSettings())) {
+      SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
+      GapicUnbufferedChunkedResumableWritableByteChannel c =
+          new GapicUnbufferedChunkedResumableWritableByteChannel(
+              result,
+              segmenter,
+              sc.writeObjectCallable(),
+              reqFactory,
+              RetryingDependencies.attemptOnce(),
+              Retrying.neverRetry(),
+              Retrying::newCallContext);
+      try {
+        int written = c.writeAndClose(ByteBuffer.wrap(bytes));
+        assertThat(written).isEqualTo(bytes.length);
+      } catch (PermissionDeniedException ignore) {
+      }
+      assertThat(result.get()).isEqualTo(resp5);
+    }
   }
 
   static class DirectWriteService extends StorageImplBase {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
deleted file mode 100644
index f4c037b1..00000000
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * Copyright 2022 Google LLC
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import static com.google.common.truth.Truth.assertThat;
-
-import com.google.api.gax.grpc.GrpcCallContext;
-import com.google.api.gax.rpc.ApiCallContext;
-import com.google.api.gax.rpc.ApiStreamObserver;
-import com.google.api.gax.rpc.ClientStreamingCallable;
-import com.google.cloud.storage.Retrying.RetryingDependencies;
-import com.google.cloud.storage.WriteFlushStrategy.Flusher;
-import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.storage.v2.WriteObjectRequest;
-import com.google.storage.v2.WriteObjectResponse;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Function;
-import org.junit.Test;
-
-public final class WriteFlushStrategyTest {
-  private static final Map<String, List<String>> expectedHeaderNonNullNonEmpty =
-      ImmutableMap.of("x-goog-request-params", ImmutableList.of("bucket=bucket-name"));
-  private static final Map<String, List<String>> expectedHeaderNonNullEmpty = ImmutableMap.of();
-  private static final Map<String, List<String>> expectedHeaderNull = ImmutableMap.of();
-
-  @Test
-  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncEveryFlush() {
-    doTest(
-        write ->
-            WriteFlushStrategy.fsyncEveryFlush(
-                write,
-                RetryingDependencies.attemptOnce(),
-                Retrying.neverRetry(),
-                GrpcCallContext::createDefault),
-        "bucket-name",
-        expectedHeaderNonNullNonEmpty);
-  }
-
-  @Test
-  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncOnClose() {
-    doTest(WriteFlushStrategy::fsyncOnClose, "bucket-name", expectedHeaderNonNullNonEmpty);
-  }
-
-  @Test
-  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncEveryFlush() {
-    doTest(
-        write ->
-            WriteFlushStrategy.fsyncEveryFlush(
-                write,
-                RetryingDependencies.attemptOnce(),
-                Retrying.neverRetry(),
-                GrpcCallContext::createDefault),
-        "",
-        expectedHeaderNonNullEmpty);
-  }
-
-  @Test
-  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncOnClose() {
-    doTest(WriteFlushStrategy::fsyncOnClose, "", expectedHeaderNonNullEmpty);
-  }
-
-  @Test
-  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncEveryFlush() {
-    doTest(
-        write ->
-            WriteFlushStrategy.fsyncEveryFlush(
-                write,
-                RetryingDependencies.attemptOnce(),
-                Retrying.neverRetry(),
-                GrpcCallContext::createDefault),
-        null,
-        expectedHeaderNull);
-  }
-
-  @Test
-  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncOnClose() {
-    doTest(WriteFlushStrategy::fsyncOnClose, null, expectedHeaderNull);
-  }
-
-  private static void doTest(
-      Function<ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>, FlusherFactory> ff,
-      String bucketName,
-      Map<String, List<String>> expectedHeader) {
-    AtomicLong c = new AtomicLong(0);
-    AtomicReference<WriteObjectResponse> ref = new AtomicReference<>();
-    AtomicReference<Map<String, List<String>>> actualHeader = new AtomicReference<>();
-    ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
-        new ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>() {
-          @Override
-          public ApiStreamObserver<WriteObjectRequest> clientStreamingCall(
-              ApiStreamObserver<WriteObjectResponse> responseObserver, ApiCallContext context) {
-            Map<String, List<String>> extraHeaders = context.getExtraHeaders();
-            actualHeader.compareAndSet(null, extraHeaders);
-            return new ApiStreamObserver<WriteObjectRequest>() {
-              @Override
-              public void onNext(WriteObjectRequest value) {}
-
-              @Override
-              public void onError(Throwable t) {}
-
-              @Override
-              public void onCompleted() {
-                responseObserver.onCompleted();
-              }
-            };
-          }
-        };
-    FlusherFactory factory = ff.apply(write);
-    Flusher flusher = factory.newFlusher(bucketName, c::addAndGet, ref::set);
-    flusher.flush(Collections.emptyList());
-    flusher.close(null);
-    assertThat(actualHeader.get()).isEqualTo(expectedHeader);
-  }
-}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
index 12bbf3df..a151ca03 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java
@@ -23,11 +23,14 @@ import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.TmpFile;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.ITObjectChecksumSupportTest.ChecksummedTestContentProvider;
@@ -47,8 +50,10 @@ import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -75,6 +80,7 @@ public final class ITObjectChecksumSupportTest {
     @Override
     public ImmutableList<?> parameters() {
       DataGenerator gen = DataGenerator.base64Characters();
+      int _256KiB = 256 * 1024;
       int _2MiB = 2 * 1024 * 1024;
       int _24MiB = 24 * 1024 * 1024;
 
@@ -84,7 +90,9 @@ public final class ITObjectChecksumSupportTest {
           // med, multiple messages single stream when resumable
           ChecksummedTestContent.of(gen.genBytes(_2MiB + 3)),
           // large, multiple messages and multiple streams when resumable
-          ChecksummedTestContent.of(gen.genBytes(_24MiB + 5)));
+          ChecksummedTestContent.of(gen.genBytes(_24MiB + 5)),
+          // quantum aligned number of bytes
+          ChecksummedTestContent.of(gen.genBytes(_2MiB * 8 + _256KiB)));
     }
   }
 
@@ -282,4 +290,67 @@ public final class ITObjectChecksumSupportTest {
     Blob blob = storage.get(blobId);
     assertThat(blob.getMd5()).isEqualTo(content.getMd5Base64());
   }
+
+  @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
+  public void testCrc32cValidated_bidiWrite_expectSuccess() throws Exception {
+    String blobName = generator.randomObjectName();
+    BlobId blobId = BlobId.of(bucket.getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
+
+    byte[] bytes = content.getBytes();
+
+    StorageOptions optionsWithBidi =
+        this.storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
+
+    try (Storage storage = optionsWithBidi.getService()) {
+      BlobWriteSession session =
+          storage.blobWriteSession(
+              blobInfo, BlobWriteOption.doesNotExist(), BlobWriteOption.crc32cMatch());
+
+      try (ReadableByteChannel src = Channels.newChannel(new ByteArrayInputStream(bytes));
+          WritableByteChannel dst = session.open()) {
+        ByteStreams.copy(src, dst);
+      }
+
+      BlobInfo gen1 = session.getResult().get(5, TimeUnit.SECONDS);
+      assertThat(gen1.getCrc32c()).isEqualTo(content.getCrc32cBase64());
+    }
+  }
+
+  @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
+  public void testCrc32cValidated_bidiWrite_expectFailure() throws Exception {
+    String blobName = generator.randomObjectName();
+    BlobId blobId = BlobId.of(bucket.getName(), blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build();
+
+    byte[] bytes = content.concat('x');
+
+    StorageOptions optionsWithBidi =
+        this.storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
+
+    try (Storage storage = optionsWithBidi.getService()) {
+      BlobWriteSession session =
+          storage.blobWriteSession(
+              blobInfo, BlobWriteOption.doesNotExist(), BlobWriteOption.crc32cMatch());
+
+      WritableByteChannel dst = session.open();
+      try (ReadableByteChannel src = Channels.newChannel(new ByteArrayInputStream(bytes))) {
+        ByteStreams.copy(src, dst);
+      }
+
+      StorageException expected = assertThrows(StorageException.class, dst::close);
+
+      assertThat(expected.getCode()).isEqualTo(400);
+    }
+  }
 }
diff --git a/samples/snippets/src/test/java/com/example/storage/transfermanager/ITTransferManagerSamples.java b/samples/snippets/src/test/java/com/example/storage/transfermanager/ITTransferManagerSamples.java
index 4a9a3d04..efc41086 100644
--- a/samples/snippets/src/test/java/com/example/storage/transfermanager/ITTransferManagerSamples.java
+++ b/samples/snippets/src/test/java/com/example/storage/transfermanager/ITTransferManagerSamples.java
@@ -25,11 +25,19 @@ import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.cloud.testing.junit4.StdOutCaptureRule;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import java.io.File;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.OpenOption;
 import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
@@ -39,6 +47,7 @@ public class ITTransferManagerSamples {
   private static final String BUCKET = RemoteStorageHelper.generateBucketName();
   private static Storage storage;
   private static List<BlobInfo> blobs;
+  private static List<BlobInfo> bigBlob;
   private static final String PROJECT_ID = System.getenv("GOOGLE_CLOUD_PROJECT");
   @Rule public final StdOutCaptureRule stdOutCaptureRule = new StdOutCaptureRule();
   @Rule public final TemporaryFolder tmp = new TemporaryFolder();
@@ -112,4 +121,23 @@ public class ITTransferManagerSamples {
     assertThat(snippetOutput.contains("blob2")).isTrue();
     assertThat(snippetOutput.contains("blob3")).isTrue();
   }
+
+  @Test
+  public void uploadAllowPCU() throws IOException {
+    File tmpFile = tmpDirectory.newFile("fileDirUpload.txt");
+    AllowParallelCompositeUpload
+        .parallelCompositeUploadAllowed(BUCKET, Collections.singletonList(tmpFile.toPath()));
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("fileDirUpload.txt")).isTrue();
+  }
+
+  @Test
+  public void downloadAllowDivideAndConquer() {
+    AllowDivideAndConquerDownload
+        .divideAndConquerDownloadAllowed(blobs, BUCKET, tmp.getRoot().toPath());
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("blob1")).isTrue();
+    assertThat(snippetOutput.contains("blob2")).isTrue();
+    assertThat(snippetOutput.contains("blob3")).isTrue();
+  }
 }