Repository: googleapis/java-storage
Pull request #2343 (2024-01-02 19:14:32)

feat: Add Bidi write feature
This adds a new BlobWriteSessionConfig which uses the Bidi streaming feature. It's largely the same as our normal resumable upload flow, with two key differences:

-Instead of closing and re-opening a stream on each flush, we keep a stream open for the life of the upload
-On each flush, it sends a signal to GCS with the "flush" and "state_lookup" flags (which are exclusive to bidi writes) set to true

I still need to run a formal profiling test, but basic profiling in IntelliJ is consistent with the 10-15% increase we're expecting


Patch:
diff --git a/.kokoro/presubmit/graalvm-native-17.cfg b/.kokoro/presubmit/graalvm-native-17.cfg
index ce36c4a9..3ab6b841 100644
--- a/.kokoro/presubmit/graalvm-native-17.cfg
+++ b/.kokoro/presubmit/graalvm-native-17.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-kokoro-resources/graalvm17:22.3.3"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.25.0"
 }
 
 env_vars: {
diff --git a/.kokoro/presubmit/graalvm-native.cfg b/.kokoro/presubmit/graalvm-native.cfg
index ee6ff821..7cca24aa 100644
--- a/.kokoro/presubmit/graalvm-native.cfg
+++ b/.kokoro/presubmit/graalvm-native.cfg
@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-kokoro-resources/graalvm:22.3.3"
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.25.0"
 }
 
 env_vars: {
diff --git a/README.md b/README.md
index 269e3c08..038279f8 100644
--- a/README.md
+++ b/README.md
@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-storage'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-storage:2.33.0'
+implementation 'com.google.cloud:google-cloud-storage:2.34.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.33.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.34.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -428,7 +428,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.33.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.34.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles
diff --git a/google-cloud-storage-bom/pom.xml b/google-cloud-storage-bom/pom.xml
index 6282caa3..94932f9e 100644
--- a/google-cloud-storage-bom/pom.xml
+++ b/google-cloud-storage-bom/pom.xml
@@ -23,8 +23,8 @@
   <packaging>pom</packaging>
   <parent>
     <groupId>com.google.cloud</groupId>
-    <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.7.1</version>
+    <artifactId>sdk-platform-java-config</artifactId>
+    <version>3.25.0</version>
     <relativePath/>
   </parent>
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
new file mode 100644
index 00000000..06dd7e7a
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
@@ -0,0 +1,180 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.BidiWriteObjectResponse;
+import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
+import java.time.Clock;
+
+public class BidiBlobWriteSessionConfig extends BlobWriteSessionConfig
+    implements BlobWriteSessionConfig.GrpcCompatible {
+  private static final long serialVersionUID = -903533790705476197L;
+
+  private final int bufferSize;
+
+  @InternalApi
+  BidiBlobWriteSessionConfig(int bufferSize) {
+    this.bufferSize = bufferSize;
+  }
+
+  /**
+   * The number of bytes to hold in the buffer before each flush
+   *
+   * <p><i>Default:</i> {@code 16777216 (16 MiB)}
+   *
+   * @see #withBufferSize(int)
+   * @since 2.34.0 This new api is in preview and is subject to breaking changes.
+   */
+  public int getBufferSize() {
+    return bufferSize;
+  }
+
+  @Override
+  WriterFactory createFactory(Clock clock) throws IOException {
+    return new Factory(ByteSizeConstants._16MiB);
+  }
+
+  @InternalApi
+  private static final class Factory implements WriterFactory {
+    private static final Conversions.Decoder<BidiWriteObjectResponse, BlobInfo>
+        WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
+            Conversions.grpc().blobInfo().compose(BidiWriteObjectResponse::getResource);
+
+    private final int bufferSize;
+
+    private Factory(int bufferSize) {
+      this.bufferSize = bufferSize;
+    }
+
+    @InternalApi
+    @Override
+    public WritableByteChannelSession<?, BlobInfo> writeSession(
+        StorageInternal s, BlobInfo info, UnifiedOpts.Opts<UnifiedOpts.ObjectTargetOpt> opts) {
+      if (s instanceof GrpcStorageImpl) {
+        return new DecoratedWritableByteChannelSession<>(
+            new LazySession<>(
+                new LazyWriteChannel<>(
+                    () -> {
+                      GrpcStorageImpl grpc = (GrpcStorageImpl) s;
+                      GrpcCallContext grpcCallContext =
+                          opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+                      BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
+
+                      ApiFuture<BidiResumableWrite> startResumableWrite =
+                          grpc.startResumableWrite(grpcCallContext, req);
+                      return ResumableMedia.gapic()
+                          .write()
+                          .bidiByteChannel(grpc.storageClient.bidiWriteObjectCallable())
+                          .setHasher(Hasher.noop())
+                          .setByteStringStrategy(ByteStringStrategy.copy())
+                          .resumable()
+                          .withRetryConfig(
+                              grpc.getOptions(), grpc.retryAlgorithmManager.idempotent())
+                          .buffered(BufferHandle.allocate(bufferSize))
+                          .setStartAsync(startResumableWrite)
+                          .build();
+                    })),
+            WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
+      } else {
+        throw new IllegalStateException(
+            "Unknown Storage implementation: " + s.getClass().getName());
+      }
+    }
+  }
+
+  /**
+   * Create a new instance with the {@code bufferSize} set to the specified value.
+   *
+   * <p><i>Default:</i> {@code 16777216 (16 MiB)}
+   *
+   * @param bufferSize The number of bytes to hold in the buffer before each flush. Must be &gt;=
+   *     {@code 262144 (256 KiB)}
+   * @return The new instance
+   * @see #getBufferSize()
+   * @since 2.34.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  public BidiBlobWriteSessionConfig withBufferSize(int bufferSize) {
+    Preconditions.checkArgument(
+        bufferSize >= ByteSizeConstants._256KiB,
+        "bufferSize must be >= %d",
+        ByteSizeConstants._256KiB);
+    return new BidiBlobWriteSessionConfig(bufferSize);
+  }
+
+  private static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>
+      implements WritableByteChannelSession<WBC, BlobInfo> {
+
+    private final WritableByteChannelSession<WBC, T> delegate;
+    private final Conversions.Decoder<T, BlobInfo> decoder;
+
+    private DecoratedWritableByteChannelSession(
+        WritableByteChannelSession<WBC, T> delegate, Conversions.Decoder<T, BlobInfo> decoder) {
+      this.delegate = delegate;
+      this.decoder = decoder;
+    }
+
+    @Override
+    public WBC open() {
+      try {
+        return WritableByteChannelSession.super.open();
+      } catch (Exception e) {
+        throw StorageException.coalesce(e);
+      }
+    }
+
+    @Override
+    public ApiFuture<WBC> openAsync() {
+      return delegate.openAsync();
+    }
+
+    @Override
+    public ApiFuture<BlobInfo> getResult() {
+      return ApiFutures.transform(
+          delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
+    }
+  }
+
+  private static final class LazySession<R>
+      implements WritableByteChannelSession<
+          BufferedWritableByteChannelSession.BufferedWritableByteChannel, R> {
+    private final LazyWriteChannel<R> lazy;
+
+    private LazySession(LazyWriteChannel<R> lazy) {
+      this.lazy = lazy;
+    }
+
+    @Override
+    public ApiFuture<BufferedWritableByteChannelSession.BufferedWritableByteChannel> openAsync() {
+      return lazy.getSession().openAsync();
+    }
+
+    @Override
+    public ApiFuture<R> getResult() {
+      return lazy.getSession().getResult();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java
new file mode 100644
index 00000000..dab7b247
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2023 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.StorageV2ProtoUtils.fmtProto;
+
+import com.google.cloud.storage.BidiWriteCtx.BidiWriteObjectRequestBuilderFactory;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import java.util.Objects;
+import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class BidiResumableWrite implements BidiWriteObjectRequestBuilderFactory {
+
+  private final StartResumableWriteRequest req;
+  private final StartResumableWriteResponse res;
+
+  private final BidiWriteObjectRequest writeRequest;
+
+  public BidiResumableWrite(
+      StartResumableWriteRequest req,
+      StartResumableWriteResponse res,
+      Function<String, BidiWriteObjectRequest> f) {
+    this.req = req;
+    this.res = res;
+    this.writeRequest = f.apply(res.getUploadId());
+  }
+
+  public StartResumableWriteRequest getReq() {
+    return req;
+  }
+
+  public StartResumableWriteResponse getRes() {
+    return res;
+  }
+
+  @Override
+  public BidiWriteObjectRequest.Builder newBuilder() {
+    return writeRequest.toBuilder();
+  }
+
+  @Override
+  public @Nullable String bucketName() {
+    if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+      return req.getWriteObjectSpec().getResource().getBucket();
+    }
+    return null;
+  }
+
+  @Override
+  public String toString() {
+    return "BidiResumableWrite{" + "req=" + fmtProto(req) + ", res=" + fmtProto(res) + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof ResumableWrite)) {
+      return false;
+    }
+    ResumableWrite resumableWrite = (ResumableWrite) o;
+    return Objects.equals(req, resumableWrite.getReq())
+        && Objects.equals(res, resumableWrite.getRes());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(req, res);
+  }
+
+  /**
+   * Helper function which is more specific than {@link Function#identity()}. Constraining the input
+   * and output to be exactly {@link BidiResumableWrite}.
+   */
+  static BidiResumableWrite identity(BidiResumableWrite w) {
+    return w;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java
new file mode 100644
index 00000000..09e4177c
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiWriteCtx.java
@@ -0,0 +1,119 @@
+/*
+ * Copyright 2023 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.StorageV2ProtoUtils.fmtProto;
+
+import com.google.cloud.storage.BidiWriteCtx.BidiWriteObjectRequestBuilderFactory;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class BidiWriteCtx<RequestFactoryT extends BidiWriteObjectRequestBuilderFactory> {
+
+  private final RequestFactoryT requestFactory;
+
+  private final AtomicLong totalSentBytes;
+  private final AtomicLong confirmedBytes;
+  private final AtomicReference<Crc32cLengthKnown> cumulativeCrc32c;
+
+  BidiWriteCtx(RequestFactoryT requestFactory) {
+    this.requestFactory = requestFactory;
+    this.totalSentBytes = new AtomicLong(0);
+    this.confirmedBytes = new AtomicLong(0);
+    this.cumulativeCrc32c = new AtomicReference<>();
+  }
+
+  public RequestFactoryT getRequestFactory() {
+    return requestFactory;
+  }
+
+  public BidiWriteObjectRequest.Builder newRequestBuilder() {
+    return requestFactory.newBuilder();
+  }
+
+  public AtomicLong getTotalSentBytes() {
+    return totalSentBytes;
+  }
+
+  public AtomicLong getConfirmedBytes() {
+    return confirmedBytes;
+  }
+
+  public AtomicReference<Crc32cLengthKnown> getCumulativeCrc32c() {
+    return cumulativeCrc32c;
+  }
+
+  // TODO: flush this out more
+  boolean isDirty() {
+    return confirmedBytes.get() == totalSentBytes.get();
+  }
+
+  @Override
+  public String toString() {
+    return "ServerState{"
+        + "requestFactory="
+        + requestFactory
+        + ", totalSentBytes="
+        + totalSentBytes
+        + ", confirmedBytes="
+        + confirmedBytes
+        + ", totalSentCrc32c="
+        + cumulativeCrc32c
+        + '}';
+  }
+
+  interface BidiWriteObjectRequestBuilderFactory {
+    BidiWriteObjectRequest.Builder newBuilder();
+
+    @Nullable
+    String bucketName();
+
+    static BidiSimpleWriteObjectRequestBuilderFactory simple(BidiWriteObjectRequest req) {
+      return new BidiSimpleWriteObjectRequestBuilderFactory(req);
+    }
+  }
+
+  static final class BidiSimpleWriteObjectRequestBuilderFactory
+      implements BidiWriteObjectRequestBuilderFactory {
+    private final BidiWriteObjectRequest req;
+
+    private BidiSimpleWriteObjectRequestBuilderFactory(BidiWriteObjectRequest req) {
+      this.req = req;
+    }
+
+    @Override
+    public BidiWriteObjectRequest.Builder newBuilder() {
+      return req.toBuilder();
+    }
+
+    @Override
+    public @Nullable String bucketName() {
+      if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+        return req.getWriteObjectSpec().getResource().getBucket();
+      }
+      return null;
+    }
+
+    @Override
+    public String toString() {
+      return "SimpleBidiWriteObjectRequestBuilderFactory{" + "req=" + fmtProto(req) + '}';
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
index 0f11cb19..43b68d8a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
@@ -246,6 +246,21 @@ public final class BlobWriteSessionConfigs {
     return new DefaultBlobWriteSessionConfig(ByteSizeConstants._16MiB);
   }
 
+  /**
+   * Factory to produce a resumable upload using a bi-directional stream. This should provide a
+   * small performance increase compared to a regular resumable upload.
+   *
+   * <p>Configuration of the buffer size can be performed via {@link
+   * BidiBlobWriteSessionConfig#withBufferSize(int)}.
+   *
+   * @since 2.34.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  @TransportCompatibility({Transport.GRPC})
+  public static BidiBlobWriteSessionConfig bidiWrite() {
+    return new BidiBlobWriteSessionConfig(ByteSizeConstants._16MiB);
+  }
+
   /**
    * Create a new {@link BlobWriteSessionConfig} which will first buffer the content of the object
    * to a temporary file under {@code java.io.tmpdir}.
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
new file mode 100644
index 00000000..d19a8802
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java
@@ -0,0 +1,165 @@
+/*
+ * Copyright 2023 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.cloud.storage.ChunkSegmenter.ChunkSegment;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.BidiWriteObjectResponse;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.ObjectChecksums;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.List;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+final class GapicBidiUnbufferedWritableByteChannel<
+        RequestFactoryT extends BidiWriteCtx.BidiWriteObjectRequestBuilderFactory>
+    implements UnbufferedWritableByteChannel {
+
+  private final SettableApiFuture<BidiWriteObjectResponse> resultFuture;
+  private final ChunkSegmenter chunkSegmenter;
+
+  private final BidiWriteCtx<RequestFactoryT> writeCtx;
+  private final WriteFlushStrategy.BidiFlusher flusher;
+
+  private boolean open = true;
+  private boolean finished = false;
+
+  GapicBidiUnbufferedWritableByteChannel(
+      SettableApiFuture<BidiWriteObjectResponse> resultFuture,
+      ChunkSegmenter chunkSegmenter,
+      RequestFactoryT requestFactory,
+      WriteFlushStrategy.BidiFlusherFactory flusherFactory) {
+    this.resultFuture = resultFuture;
+    this.chunkSegmenter = chunkSegmenter;
+
+    this.writeCtx = new BidiWriteCtx<>(requestFactory);
+    this.flusher =
+        flusherFactory.newFlusher(
+            requestFactory.bucketName(), 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 boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (!finished) {
+      BidiWriteObjectRequest message = finishMessage();
+      try {
+        flusher.close(message);
+        finished = true;
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
+    } else {
+      flusher.close(null);
+    }
+    open = false;
+  }
+
+  @VisibleForTesting
+  BidiWriteCtx<RequestFactoryT> getWriteCtx() {
+    return writeCtx;
+  }
+
+  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<BidiWriteObjectRequest> messages = new ArrayList<>();
+
+    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());
+      }
+      BidiWriteObjectRequest.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;
+      }
+
+      BidiWriteObjectRequest build = builder.build();
+      messages.add(build);
+      bytesConsumed += contentSize;
+    }
+    if (finalize && !finished) {
+      messages.add(finishMessage());
+      finished = true;
+    }
+
+    try {
+      flusher.flush(messages);
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
+    }
+
+    return bytesConsumed;
+  }
+
+  @NonNull
+  private BidiWriteObjectRequest finishMessage() {
+    long offset = writeCtx.getTotalSentBytes().get();
+    Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+
+    BidiWriteObjectRequest.Builder b =
+        writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+    if (crc32cValue != null) {
+      b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
+    }
+    BidiWriteObjectRequest message = b.build();
+    return message;
+  }
+}
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
new file mode 100644
index 00000000..e26d33bb
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java
@@ -0,0 +1,178 @@
+/*
+ * Copyright 2023 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 java.util.Objects.requireNonNull;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.rpc.BidiStreamingCallable;
+import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.storage.v2.BidiWriteObjectRequest;
+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 {
+
+  private final BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write;
+  private Hasher hasher;
+  private ByteStringStrategy byteStringStrategy;
+
+  GapicBidiWritableByteChannelSessionBuilder(
+      BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write) {
+    this.write = write;
+    this.hasher = Hasher.noop();
+    this.byteStringStrategy = ByteStringStrategy.copy();
+  }
+
+  /**
+   * Set the {@link Hasher} to apply to the bytes passing through the built session's channel.
+   *
+   * <p>Default: {@link Hasher#noop()}
+   *
+   * @see Hasher#enabled()
+   * @see Hasher#noop()
+   */
+  GapicBidiWritableByteChannelSessionBuilder setHasher(Hasher hasher) {
+    this.hasher = requireNonNull(hasher, "hasher must be non null");
+    return this;
+  }
+
+  /**
+   * Set the {@link ByteStringStrategy} to be used when constructing {@link
+   * com.google.protobuf.ByteString ByteString}s from {@link ByteBuffer}s.
+   *
+   * <p>Default: {@link ByteStringStrategy#copy()}
+   *
+   * <p>Note: usage of {@link ByteStringStrategy#noCopy()} requires that any {@link ByteBuffer}
+   * passed to the session's channel not be modified while {@link
+   * java.nio.channels.WritableByteChannel#write(ByteBuffer)} is processing.
+   *
+   * @see ByteStringStrategy#copy()
+   * @see ByteStringStrategy#noCopy()
+   */
+  GapicBidiWritableByteChannelSessionBuilder setByteStringStrategy(
+      ByteStringStrategy byteStringStrategy) {
+    this.byteStringStrategy =
+        requireNonNull(byteStringStrategy, "byteStringStrategy must be non null");
+    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();
+  }
+
+  final class ResumableUploadBuilder {
+
+    private RetryingDependencies deps;
+    private ResultRetryAlgorithm<?> alg;
+
+    ResumableUploadBuilder() {
+      this.deps = RetryingDependencies.attemptOnce();
+      this.alg = Retrying.neverRetry();
+    }
+
+    ResumableUploadBuilder withRetryConfig(RetryingDependencies deps, ResultRetryAlgorithm<?> alg) {
+      this.deps = requireNonNull(deps, "deps must be non null");
+      this.alg = requireNonNull(alg, "alg must be non null");
+      return this;
+    }
+
+    /**
+     * Buffer using {@code byteBuffer} worth of space before attempting to flush.
+     *
+     * <p>The provided {@link ByteBuffer} <i>should</i> be aligned with GCSs block size of <a
+     * target="_blank"
+     * href="https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload">256
+     * KiB</a>.
+     */
+    BufferedResumableUploadBuilder buffered(ByteBuffer byteBuffer) {
+      return buffered(BufferHandle.handleOf(byteBuffer));
+    }
+
+    BufferedResumableUploadBuilder buffered(BufferHandle bufferHandle) {
+      return new BufferedResumableUploadBuilder(bufferHandle);
+    }
+
+    final class BufferedResumableUploadBuilder {
+
+      private final BufferHandle bufferHandle;
+
+      private ApiFuture<BidiResumableWrite> start;
+
+      BufferedResumableUploadBuilder(BufferHandle bufferHandle) {
+        this.bufferHandle = bufferHandle;
+      }
+
+      /**
+       * Set the Future which will contain the ResumableWrite information necessary to open the
+       * Write stream.
+       */
+      BufferedResumableUploadBuilder setStartAsync(ApiFuture<BidiResumableWrite> start) {
+        this.start = requireNonNull(start, "start must be non null");
+        return this;
+      }
+
+      BufferedWritableByteChannelSession<BidiWriteObjectResponse> build() {
+        return new BufferedWriteSession<>(
+            requireNonNull(start, "start must be non null"),
+            bindFunction(
+                    WriteFlushStrategy.defaultBidiFlusher(
+                        write, deps, alg, Retrying::newCallContext),
+                    BidiResumableWrite::identity)
+                .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
index 7b4c6a94..ac520213 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
@@ -18,9 +18,12 @@ package com.google.cloud.storage;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.gax.rpc.BidiStreamingCallable;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.BidiWriteObjectResponse;
 import com.google.storage.v2.StartResumableWriteRequest;
 import com.google.storage.v2.StartResumableWriteResponse;
 import com.google.storage.v2.WriteObjectRequest;
@@ -40,6 +43,11 @@ final class GapicUploadSessionBuilder {
     return new GapicWritableByteChannelSessionBuilder(write);
   }
 
+  GapicBidiWritableByteChannelSessionBuilder bidiByteChannel(
+      BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse> write) {
+    return new GapicBidiWritableByteChannelSessionBuilder(write);
+  }
+
   ApiFuture<ResumableWrite> resumableWrite(
       UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
       WriteObjectRequest writeObjectRequest) {
@@ -62,4 +70,27 @@ final class GapicUploadSessionBuilder {
         (resp) -> new ResumableWrite(req, resp, f),
         MoreExecutors.directExecutor());
   }
+
+  ApiFuture<BidiResumableWrite> bidiResumableWrite(
+      UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
+      BidiWriteObjectRequest writeObjectRequest) {
+    StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
+    if (writeObjectRequest.hasWriteObjectSpec()) {
+      b.setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec());
+    }
+    if (writeObjectRequest.hasCommonObjectRequestParams()) {
+      b.setCommonObjectRequestParams(writeObjectRequest.getCommonObjectRequestParams());
+    }
+    if (writeObjectRequest.hasObjectChecksums()) {
+      b.setObjectChecksums(writeObjectRequest.getObjectChecksums());
+    }
+    StartResumableWriteRequest req = b.build();
+    Function<String, BidiWriteObjectRequest> f =
+        uploadId ->
+            writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();
+    return ApiFutures.transform(
+        x.futureCall(req),
+        (resp) -> new BidiResumableWrite(req, resp, f),
+        MoreExecutors.directExecutor());
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
index 703cd152..ea481ab2 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcRetryAlgorithmManager.java
@@ -20,6 +20,7 @@ import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
+import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.ComposeObjectRequest;
 import com.google.storage.v2.CreateBucketRequest;
 import com.google.storage.v2.CreateHmacKeyRequest;
@@ -201,4 +202,10 @@ final class GrpcRetryAlgorithmManager implements Serializable {
         ? retryStrategy.getIdempotentHandler()
         : retryStrategy.getNonidempotentHandler();
   }
+
+  public ResultRetryAlgorithm<?> getFor(BidiWriteObjectRequest req) {
+    return req.getWriteObjectSpec().hasIfGenerationMatch()
+        ? retryStrategy.getIdempotentHandler()
+        : retryStrategy.getNonidempotentHandler();
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index 94ed6b33..53c73d4d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -84,6 +84,7 @@ import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.FieldMask;
+import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BucketAccessControl;
 import com.google.storage.v2.ComposeObjectRequest;
 import com.google.storage.v2.ComposeObjectRequest.SourceObject;
@@ -1807,8 +1808,30 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     return opts.writeObjectRequest().apply(requestBuilder).build();
   }
 
+  BidiWriteObjectRequest getBidiWriteObjectRequest(BlobInfo info, Opts<ObjectTargetOpt> opts) {
+    Object object = codecs.blobInfo().encode(info);
+    Object.Builder objectBuilder =
+        object
+            .toBuilder()
+            // required if the data is changing
+            .clearChecksums()
+            // trimmed to shave payload size
+            .clearGeneration()
+            .clearMetageneration()
+            .clearSize()
+            .clearCreateTime()
+            .clearUpdateTime();
+    WriteObjectSpec.Builder specBuilder = WriteObjectSpec.newBuilder().setResource(objectBuilder);
+
+    BidiWriteObjectRequest.Builder requestBuilder =
+        BidiWriteObjectRequest.newBuilder().setWriteObjectSpec(specBuilder);
+
+    return opts.bidiWriteObjectRequest().apply(requestBuilder).build();
+  }
+
   private UnbufferedReadableByteChannelSession<Object> unbufferedReadSession(
       BlobId blob, BlobSourceOption[] options) {
+
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes =
@@ -1838,6 +1861,19 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
             req);
   }
 
+  ApiFuture<BidiResumableWrite> startResumableWrite(
+      GrpcCallContext grpcCallContext, BidiWriteObjectRequest req) {
+    Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
+    GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
+    return ResumableMedia.gapic()
+        .write()
+        .bidiResumableWrite(
+            storageClient
+                .startResumableWriteCallable()
+                .withDefaultCallContext(merge.withRetryableCodes(codes)),
+            req);
+  }
+
   private SourceObject sourceObjectEncode(SourceBlob from) {
     SourceObject.Builder to = SourceObject.newBuilder();
     to.setName(from.getName());
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index 18f79be5..dbf7bb64 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -37,6 +37,7 @@ import com.google.common.io.BaseEncoding;
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.FieldMask;
+import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.CommonObjectRequestParams;
 import com.google.storage.v2.ComposeObjectRequest;
 import com.google.storage.v2.CreateBucketRequest;
@@ -171,6 +172,10 @@ final class UnifiedOpts {
       return Mapper.identity();
     }
 
+    default Mapper<BidiWriteObjectRequest.Builder> bidiWriteObject() {
+      return Mapper.identity();
+    }
+
     default Mapper<UpdateObjectRequest.Builder> updateObject() {
       return Mapper.identity();
     }
@@ -2266,6 +2271,10 @@ final class UnifiedOpts {
       return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::writeObject);
     }
 
+    Mapper<BidiWriteObjectRequest.Builder> bidiWriteObjectRequest() {
+      return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::bidiWriteObject);
+    }
+
     Mapper<GetObjectRequest.Builder> getObjectsRequest() {
       return fuseMappers(ObjectSourceOpt.class, ObjectSourceOpt::getObject);
     }
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
index 806c2b42..0a191c7d 100644
--- 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
@@ -20,11 +20,14 @@ 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;
@@ -72,6 +75,28 @@ final class WriteFlushStrategy {
             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.
@@ -121,6 +146,27 @@ final class WriteFlushStrategy {
     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 {
     /**
@@ -140,6 +186,25 @@ final class WriteFlushStrategy {
     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;
@@ -199,6 +264,82 @@ final class WriteFlushStrategy {
     }
   }
 
+  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;
@@ -315,4 +456,66 @@ final class WriteFlushStrategy {
       }
     }
   }
+
+  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/pom.xml b/pom.xml
index 9de1f695..1e724000 100644
--- a/pom.xml
+++ b/pom.xml
@@ -13,8 +13,8 @@
 
   <parent>
     <groupId>com.google.cloud</groupId>
-    <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.7.1</version>
+    <artifactId>sdk-platform-java-config</artifactId>
+    <version>3.25.0</version>
   </parent>
 
   <developers>
@@ -62,7 +62,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>
-        <version>${google.cloud.shared-dependencies.version}</version>
+        <version>${google-cloud-shared-dependencies.version}</version>
         <type>pom</type>
         <scope>import</scope>
         <exclusions>
diff --git a/renovate.json b/renovate.json
index b5e21946..78317cac 100644
--- a/renovate.json
+++ b/renovate.json
@@ -11,6 +11,17 @@
     ":autodetectPinVersions"
   ],
   "ignorePaths": [".kokoro/requirements.txt"],
+  "customManagers": [
+    {
+      "customType": "regex",
+      "fileMatch": [
+        "^.kokoro/presubmit/graalvm-native.*.cfg$"
+      ],
+      "matchStrings": ["value: \"gcr.io/cloud-devrel-public-resources/graalvm.*:(?<currentValue>.*?)\""],
+      "depNameTemplate": "com.google.cloud:sdk-platform-java-config",
+      "datasourceTemplate": "maven"
+    }
+  ],
   "packageRules": [
     {
       "packagePatterns": [
diff --git a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/Bidi.java b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/Bidi.java
new file mode 100644
index 00000000..aa59e0b3
--- /dev/null
+++ b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/Bidi.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.benchmarking;
+
+import static com.google.cloud.storage.benchmarking.StorageSharedBenchmarkingUtils.generateCloudMonitoringResult;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import java.io.PrintWriter;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.Callable;
+
+class Bidi implements Callable<String> {
+  private final Storage storageClient;
+  private final String bucketName;
+  private final int objectSize;
+  private final PrintWriter pw;
+  private final String api;
+  private final int workers;
+
+  Bidi(
+      Storage storageClient,
+      String bucketName,
+      int objectSize,
+      PrintWriter pw,
+      String api,
+      int workers) {
+    this.storageClient = storageClient;
+    this.bucketName = bucketName;
+    this.objectSize = objectSize;
+    this.pw = pw;
+    this.api = api;
+    this.workers = workers;
+  }
+
+  @Override
+  public String call() throws Exception {
+    String blobName = DataGenerator.base64Characters().genBytes(20).toString();
+    BlobWriteSession sess =
+        storageClient.blobWriteSession(
+            BlobInfo.newBuilder(bucketName, blobName).build(), BlobWriteOption.doesNotExist());
+    byte[] bytes = DataGenerator.base64Characters().genBytes(objectSize);
+    Clock clock = Clock.systemDefaultZone();
+    Instant startTime = clock.instant();
+    try (WritableByteChannel w = sess.open()) {
+      w.write(ByteBuffer.wrap(bytes));
+    }
+    BlobInfo created = sess.getResult().get();
+    Instant endTime = clock.instant();
+    Duration elapsedTimeWrite = Duration.between(startTime, endTime);
+    printResult("BIDI", created, elapsedTimeWrite);
+    StorageSharedBenchmarkingUtils.cleanupObject(storageClient, created);
+    return "OK";
+  }
+
+  private void printResult(String op, BlobInfo created, Duration duration) {
+    pw.println(
+        generateCloudMonitoringResult(
+                op,
+                StorageSharedBenchmarkingUtils.calculateThroughput(
+                    created.getSize().doubleValue(), duration),
+                created,
+                api,
+                workers)
+            .formatAsCustomMetric());
+  }
+}
diff --git a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingCli.java b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingCli.java
index e6c4c056..836c28d0 100644
--- a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingCli.java
+++ b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingCli.java
@@ -19,6 +19,7 @@ package com.google.cloud.storage.benchmarking;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ListenableFutureToApiFuture;
 import com.google.api.gax.retrying.RetrySettings;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -106,6 +107,12 @@ public final class StorageSharedBenchmarkingCli implements Runnable {
       case "w1r3":
         runWorkload1();
         break;
+      case "bidi":
+        runWorkloadBidi();
+        break;
+      case "default-nobidi":
+        runWorkloadNoBidi();
+        break;
       default:
         throw new IllegalStateException("Specify a workload to run");
     }
@@ -151,6 +158,36 @@ public final class StorageSharedBenchmarkingCli implements Runnable {
     }
   }
 
+  private void runWorkloadBidi() {
+    StorageOptions options =
+        StorageOptions.grpc()
+            .setProjectId(project)
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
+    Storage storageClient = options.getService();
+    try {
+      runBidi(storageClient);
+    } catch (Exception e) {
+      System.err.println("Failed to run workload bidi" + e.getMessage());
+      System.exit(1);
+    }
+  }
+
+  private void runWorkloadNoBidi() {
+    StorageOptions options =
+        StorageOptions.grpc()
+            .setProjectId(project)
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.getDefault())
+            .build();
+    Storage storageClient = options.getService();
+    try {
+      runBidi(storageClient);
+    } catch (Exception e) {
+      System.err.println("Failed to run workload no bidi" + e.getMessage());
+      System.exit(1);
+    }
+  }
+
   private void runW1R3(Storage storageClient) throws ExecutionException, InterruptedException {
     ListeningExecutorService executorService =
         MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(workers));
@@ -173,6 +210,19 @@ public final class StorageSharedBenchmarkingCli implements Runnable {
     }
   }
 
+  private void runBidi(Storage storageClient) throws ExecutionException, InterruptedException {
+    ListeningExecutorService executorService =
+        MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(workers));
+    for (int i = 0; i < samples; i++) {
+      Range objectSizeRange = Range.of(objectSize);
+      int objectSize = getRandomInt(objectSizeRange.min, objectSizeRange.max);
+      convert(
+              executorService.submit(
+                  new Bidi(storageClient, bucket, objectSize, printWriter, api, workers)))
+          .get();
+    }
+  }
+
   private void runWarmup(Storage storageClient) throws ExecutionException, InterruptedException {
     if (warmup <= 0) {
       return;
diff --git a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingUtils.java b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingUtils.java
index 137fee51..52420723 100644
--- a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingUtils.java
+++ b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/StorageSharedBenchmarkingUtils.java
@@ -15,7 +15,7 @@
  */
 package com.google.cloud.storage.benchmarking;
 
-import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Storage;
 import java.time.Duration;
 
@@ -23,7 +23,7 @@ class StorageSharedBenchmarkingUtils {
   public static long SSB_SIZE_THRESHOLD_BYTES = 1048576;
   public static int DEFAULT_NUMBER_OF_READS = 3;
 
-  public static void cleanupObject(Storage storage, Blob created) {
+  public static void cleanupObject(Storage storage, BlobInfo created) {
     storage.delete(
         created.getBlobId(), Storage.BlobSourceOption.generationMatch(created.getGeneration()));
   }
@@ -36,4 +36,25 @@ class StorageSharedBenchmarkingUtils {
     double throughput = adjustedSize / (elapsedTime.toMillis() / 1000D);
     return throughput;
   }
+
+  public static CloudMonitoringResult generateCloudMonitoringResult(
+      String op, double throughput, BlobInfo created, String api, int workers) {
+    CloudMonitoringResult result =
+        CloudMonitoringResult.newBuilder()
+            .setLibrary("java")
+            .setApi(api)
+            .setOp(op)
+            .setWorkers(workers)
+            .setObjectSize(created.getSize().intValue())
+            .setChunksize(created.getSize().intValue())
+            .setCrc32cEnabled(false)
+            .setMd5Enabled(false)
+            .setCpuTimeUs(-1)
+            .setBucketName(created.getBucket())
+            .setStatus("OK")
+            .setTransferSize(created.getSize().toString())
+            .setThroughput(throughput)
+            .build();
+    return result;
+  }
 }
diff --git a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/W1R3.java b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/W1R3.java
index 83b357db..7405d9cf 100644
--- a/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/W1R3.java
+++ b/storage-shared-benchmarking/src/main/java/com/google/cloud/storage/benchmarking/W1R3.java
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage.benchmarking;
 
+import static com.google.cloud.storage.benchmarking.StorageSharedBenchmarkingUtils.generateCloudMonitoringResult;
+
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.DataGenerator;
@@ -80,7 +82,7 @@ final class W1R3 implements Callable<String> {
           printResult("READ[" + i + "]", created, elapsedTimeDownload);
         }
       }
-      StorageSharedBenchmarkingUtils.cleanupObject(storage, created);
+      StorageSharedBenchmarkingUtils.cleanupObject(storage, created.asBlobInfo());
     } catch (Exception e) {
       CloudMonitoringResult result =
           CloudMonitoringResult.newBuilder()
@@ -110,29 +112,10 @@ final class W1R3 implements Callable<String> {
                   op,
                   StorageSharedBenchmarkingUtils.calculateThroughput(
                       created.getSize().doubleValue(), duration),
-                  created)
+                  created.asBlobInfo(),
+                  api,
+                  workers)
               .formatAsCustomMetric());
     }
   }
-
-  private CloudMonitoringResult generateCloudMonitoringResult(
-      String op, double throughput, Blob created) {
-    CloudMonitoringResult result =
-        CloudMonitoringResult.newBuilder()
-            .setLibrary("java")
-            .setApi(api)
-            .setOp(op)
-            .setWorkers(workers)
-            .setObjectSize(created.getSize().intValue())
-            .setChunksize(created.getSize().intValue())
-            .setCrc32cEnabled(false)
-            .setMd5Enabled(false)
-            .setCpuTimeUs(-1)
-            .setBucketName(created.getBucket())
-            .setStatus("OK")
-            .setTransferSize(created.getSize().toString())
-            .setThroughput(throughput)
-            .build();
-    return result;
-  }
 }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
index bf8132b3..763346b1 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
@@ -102,6 +102,23 @@ public final class ITBlobWriteSessionTest {
     }
   }
 
+  @Test
+  public void bidiTest() throws Exception {
+    StorageOptions options = null;
+    if (transport == Transport.GRPC) {
+      options =
+          ((GrpcStorageOptions) storage.getOptions())
+              .toBuilder()
+              .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+              .build();
+    }
+    assertWithMessage("unable to resolve options").that(options).isNotNull();
+
+    try (Storage s = options.getService()) {
+      doTest(s);
+    }
+  }
+
   @Test
   public void closingAnOpenedSessionWithoutCallingWriteShouldMakeAnEmptyObject()
       throws IOException, ExecutionException, InterruptedException, TimeoutException {