Repository: googleapis/java-storage
Pull request #2489 (2024-04-04 18:08:02)

feat: Adds a ZeroCopy response marshaller for grpc ReadObject handling
Adds a Zero Copy ReadObject marshaller for grpc and some basic tests


Patch:
diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index d2272068..cc9d6933 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -71,4 +71,11 @@
     <className>com/google/cloud/storage/transfermanager/TransferManagerConfig$Builder</className>
     <method>* setAllowDivideAndConquer(boolean)</method>
   </difference>
+
+  <difference>
+    <differenceType>7013</differenceType>
+    <className>com/google/cloud/storage/StorageOptions$Builder</className>
+    <method>com.google.cloud.storage.StorageOptions$Builder setBlobWriteSessionConfig(com.google.cloud.storage.BlobWriteSessionConfig)</method>
+  </difference>
+
 </differences>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 7fbba0ae..c6219b77 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -96,6 +96,14 @@
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java-util</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-protobuf</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-common-protos</artifactId>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSession.java
index 02ea23a6..81cecdc8 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSession.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSession.java
@@ -20,6 +20,7 @@ import com.google.api.core.ApiFuture;
 import com.google.api.core.BetaApi;
 import java.io.IOException;
 import java.nio.channels.WritableByteChannel;
+import java.util.concurrent.TimeUnit;
 
 /**
  * A session to write an object to Google Cloud Storage.
@@ -50,6 +51,10 @@ public interface BlobWriteSession {
    * <p>Upon calling {@link WritableByteChannel#close()} the object creation will be finalized, and
    * {@link #getResult()}s future should resolve.
    *
+   * <p>The returned {@code WritableByteChannel} can throw IOExceptions from any of its usual
+   * methods. Any {@link IOException} thrown can have a cause of a {@link StorageException}.
+   * However, not all {@code IOExceptions} will have {@code StorageException}s.
+   *
    * @throws IOException When creating the {@link WritableByteChannel} if an unrecoverable
    *     underlying IOException occurs it can be rethrown
    * @throws IllegalStateException if open is called more than once
@@ -66,6 +71,10 @@ public interface BlobWriteSession {
    * Google Cloud Storage 2. A terminal failure occurs, the terminal failure will become the
    * exception result
    *
+   * <p>If a terminal failure is encountered, calling either {@link ApiFuture#get()} or {@link
+   * ApiFuture#get(long, TimeUnit)} will result in an {@link
+   * java.util.concurrent.ExecutionException} with a cause that is the {@link StorageException}.
+   *
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessions.java
index 878552a1..c9da9ee0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessions.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.ApiFuture;
+import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.nio.channels.WritableByteChannel;
 
@@ -30,14 +31,20 @@ final class BlobWriteSessions {
 
   static final class WritableByteChannelSessionAdapter implements BlobWriteSession {
     private final WritableByteChannelSession<?, BlobInfo> delegate;
+    private boolean open;
 
     private WritableByteChannelSessionAdapter(WritableByteChannelSession<?, BlobInfo> delegate) {
       this.delegate = delegate;
+      open = false;
     }
 
     @Override
     public WritableByteChannel open() throws IOException {
-      return delegate.open();
+      synchronized (this) {
+        Preconditions.checkState(!open, "already open");
+        open = true;
+        return delegate.open();
+      }
     }
 
     @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index 1d45c9e3..ffcbcfa8 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -32,6 +32,8 @@ import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
 import java.time.Clock;
 import java.util.Map;
@@ -202,24 +204,25 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
       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();
+      return ApiFutures.catchingAsync(
+          delegate.openAsync(),
+          Throwable.class,
+          throwable -> ApiFutures.immediateFailedFuture(StorageException.coalesce(throwable)),
+          MoreExecutors.directExecutor());
     }
 
     @Override
     public ApiFuture<BlobInfo> getResult() {
-      return ApiFutures.transform(
-          delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
+      ApiFuture<BlobInfo> decodeResult =
+          ApiFutures.transform(
+              delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
+      return ApiFutures.catchingAsync(
+          decodeResult,
+          Throwable.class,
+          throwable -> ApiFutures.immediateFailedFuture(StorageException.coalesce(throwable)),
+          MoreExecutors.directExecutor());
     }
   }
 
@@ -233,7 +236,55 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
 
     @Override
     public ApiFuture<BufferedWritableByteChannel> openAsync() {
-      return lazy.getSession().openAsync();
+      // make sure the errors coming out of the BufferedWritableByteChannel are either IOException
+      // or StorageException
+      return ApiFutures.transform(
+          lazy.getSession().openAsync(),
+          delegate ->
+              new BufferedWritableByteChannel() {
+                @Override
+                public int write(ByteBuffer src) throws IOException {
+                  try {
+                    return delegate.write(src);
+                  } catch (IOException e) {
+                    throw e;
+                  } catch (Exception e) {
+                    throw StorageException.coalesce(e);
+                  }
+                }
+
+                @Override
+                public void flush() throws IOException {
+                  try {
+                    delegate.flush();
+                  } catch (IOException e) {
+                    throw e;
+                  } catch (Exception e) {
+                    throw StorageException.coalesce(e);
+                  }
+                }
+
+                @Override
+                public boolean isOpen() {
+                  try {
+                    return delegate.isOpen();
+                  } catch (Exception e) {
+                    throw StorageException.coalesce(e);
+                  }
+                }
+
+                @Override
+                public void close() throws IOException {
+                  try {
+                    delegate.close();
+                  } catch (IOException e) {
+                    throw e;
+                  } catch (Exception e) {
+                    throw StorageException.coalesce(e);
+                  }
+                }
+              },
+          MoreExecutors.directExecutor());
     }
 
     @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
index 704794a3..82789bcd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
@@ -48,19 +48,23 @@ final class GapicDownloadSessionBuilder {
    * ultimately produced channel will not do any retries of its own.
    */
   public ReadableByteChannelSessionBuilder byteChannel(
-      ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read) {
-    return new ReadableByteChannelSessionBuilder(read);
+      ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
+      ResponseContentLifecycleManager responseContentLifecycleManager) {
+    return new ReadableByteChannelSessionBuilder(read, responseContentLifecycleManager);
   }
 
   public static final class ReadableByteChannelSessionBuilder {
 
     private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+    private final ResponseContentLifecycleManager responseContentLifecycleManager;
     private boolean autoGzipDecompression;
     private Hasher hasher;
 
     private ReadableByteChannelSessionBuilder(
-        ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read) {
+        ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
+        ResponseContentLifecycleManager responseContentLifecycleManager) {
       this.read = read;
+      this.responseContentLifecycleManager = responseContentLifecycleManager;
       this.hasher = Hasher.noop();
       this.autoGzipDecompression = false;
     }
@@ -100,11 +104,13 @@ final class GapicDownloadSessionBuilder {
       return (object, resultFuture) -> {
         if (autoGzipDecompression) {
           return new GzipReadableByteChannel(
-              new GapicUnbufferedReadableByteChannel(resultFuture, read, object, hasher),
+              new GapicUnbufferedReadableByteChannel(
+                  resultFuture, read, object, hasher, responseContentLifecycleManager),
               ApiFutures.transform(
                   resultFuture, Object::getContentEncoding, MoreExecutors.directExecutor()));
         } else {
-          return new GapicUnbufferedReadableByteChannel(resultFuture, read, object, hasher);
+          return new GapicUnbufferedReadableByteChannel(
+              resultFuture, read, object, hasher, responseContentLifecycleManager);
         }
       };
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
index 4b19c3f9..6cecc8dd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
@@ -16,8 +16,6 @@
 
 package com.google.cloud.storage;
 
-import static com.google.common.base.Preconditions.checkArgument;
-
 import com.google.api.client.http.HttpStatusCodes;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.SettableApiFuture;
@@ -25,6 +23,7 @@ import com.google.api.gax.rpc.ServerStream;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ReadObjectRequest;
@@ -46,6 +45,7 @@ final class GapicUnbufferedReadableByteChannel
   private final ReadObjectRequest req;
   private final Hasher hasher;
   private final LazyServerStreamIterator iter;
+  private final ResponseContentLifecycleManager rclm;
 
   private boolean open = true;
   private boolean complete = false;
@@ -53,18 +53,20 @@ final class GapicUnbufferedReadableByteChannel
 
   private Object metadata;
 
-  private ByteBuffer leftovers;
+  private ResponseContentLifecycleHandle leftovers;
 
   GapicUnbufferedReadableByteChannel(
       SettableApiFuture<Object> result,
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
       ReadObjectRequest req,
-      Hasher hasher) {
+      Hasher hasher,
+      ResponseContentLifecycleManager rclm) {
     this.result = result;
     this.read = read;
     this.req = req;
     this.hasher = hasher;
     this.blobOffset = req.getReadOffset();
+    this.rclm = rclm;
     this.iter = new LazyServerStreamIterator();
   }
 
@@ -82,8 +84,9 @@ final class GapicUnbufferedReadableByteChannel
     ReadCursor c = new ReadCursor(blobOffset, blobOffset + totalBufferCapacity);
     while (c.hasRemaining()) {
       if (leftovers != null) {
-        copy(c, leftovers, dsts, offset, length);
+        leftovers.copy(c, dsts, offset, length);
         if (!leftovers.hasRemaining()) {
+          leftovers.close();
           leftovers = null;
         }
         continue;
@@ -91,6 +94,7 @@ final class GapicUnbufferedReadableByteChannel
 
       if (iter.hasNext()) {
         ReadObjectResponse resp = iter.next();
+        ResponseContentLifecycleHandle handle = rclm.get(resp);
         if (resp.hasMetadata()) {
           Object respMetadata = resp.getMetadata();
           if (metadata == null) {
@@ -107,22 +111,24 @@ final class GapicUnbufferedReadableByteChannel
           }
         }
         ChecksummedData checksummedData = resp.getChecksummedData();
-        ByteBuffer content = checksummedData.getContent().asReadOnlyByteBuffer();
-        // very important to know whether a crc32c value is set. Without checking, protobuf will
+        ByteString content = checksummedData.getContent();
+        int contentSize = content.size();
+        // Very important to know whether a crc32c value is set. Without checking, protobuf will
         // happily return 0, which is a valid crc32c value.
         if (checksummedData.hasCrc32C()) {
-          Crc32cLengthKnown expected =
-              Crc32cValue.of(checksummedData.getCrc32C(), checksummedData.getContent().size());
+          Crc32cLengthKnown expected = Crc32cValue.of(checksummedData.getCrc32C(), contentSize);
           try {
-            hasher.validate(expected, content::duplicate);
+            hasher.validate(expected, content.asReadOnlyByteBufferList());
           } catch (IOException e) {
             close();
             throw e;
           }
         }
-        copy(c, content, dsts, offset, length);
-        if (content.hasRemaining()) {
-          leftovers = content;
+        handle.copy(c, dsts, offset, length);
+        if (handle.hasRemaining()) {
+          leftovers = handle;
+        } else {
+          handle.close();
         }
       } else {
         complete = true;
@@ -144,18 +150,19 @@ final class GapicUnbufferedReadableByteChannel
   @Override
   public void close() throws IOException {
     open = false;
-    iter.close();
+    try {
+      if (leftovers != null) {
+        leftovers.close();
+      }
+    } finally {
+      iter.close();
+    }
   }
 
   ApiFuture<Object> getResult() {
     return result;
   }
 
-  private void copy(ReadCursor c, ByteBuffer content, ByteBuffer[] dsts, int offset, int length) {
-    long copiedBytes = Buffers.copy(content, dsts, offset, length);
-    c.advance(copiedBytes);
-  }
-
   private IOException closeWithError(String message) throws IOException {
     close();
     StorageException cause =
@@ -163,40 +170,6 @@ final class GapicUnbufferedReadableByteChannel
     throw new IOException(message, cause);
   }
 
-  /**
-   * Shrink wraps a beginning, offset and limit for tracking state of an individual invocation of
-   * {@link #read}
-   */
-  private static final class ReadCursor {
-    private final long beginning;
-    private long offset;
-    private final long limit;
-
-    private ReadCursor(long beginning, long limit) {
-      this.limit = limit;
-      this.beginning = beginning;
-      this.offset = beginning;
-    }
-
-    public boolean hasRemaining() {
-      return limit - offset > 0;
-    }
-
-    public void advance(long incr) {
-      checkArgument(incr >= 0);
-      offset += incr;
-    }
-
-    public long read() {
-      return offset - beginning;
-    }
-
-    @Override
-    public String toString() {
-      return String.format("ReadCursor{begin=%d, offset=%d, limit=%d}", beginning, offset, limit);
-    }
-  }
-
   private final class LazyServerStreamIterator implements Iterator<ReadObjectResponse>, Closeable {
     private ServerStream<ReadObjectResponse> serverStream;
     private Iterator<ReadObjectResponse> responseIterator;
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/GapicUnbufferedWritableByteChannel.java
index b4a26ed4..32d5eb9d 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/GapicUnbufferedWritableByteChannel.java
@@ -92,7 +92,12 @@ final class GapicUnbufferedWritableByteChannel<
         throw e;
       }
     } else {
-      flusher.close(null);
+      try {
+        flusher.close(null);
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
     }
     open = false;
   }
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 ac520213..2d1daa54 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
@@ -49,7 +49,7 @@ final class GapicUploadSessionBuilder {
   }
 
   ApiFuture<ResumableWrite> resumableWrite(
-      UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
+      UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> callable,
       WriteObjectRequest writeObjectRequest) {
     StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
     if (writeObjectRequest.hasWriteObjectSpec()) {
@@ -65,9 +65,16 @@ final class GapicUploadSessionBuilder {
     Function<String, WriteObjectRequest> f =
         uploadId ->
             writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();
-    return ApiFutures.transform(
-        x.futureCall(req),
-        (resp) -> new ResumableWrite(req, resp, f),
+    ApiFuture<ResumableWrite> futureResumableWrite =
+        ApiFutures.transform(
+            callable.futureCall(req),
+            (resp) -> new ResumableWrite(req, resp, f),
+            MoreExecutors.directExecutor());
+    // make sure we wrap any failure as a storage exception
+    return ApiFutures.catchingAsync(
+        futureResumableWrite,
+        Throwable.class,
+        throwable -> ApiFutures.immediateFailedFuture(StorageException.coalesce(throwable)),
         MoreExecutors.directExecutor());
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
index 4ae3f244..03e3cb51 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
@@ -28,15 +28,18 @@ import org.checkerframework.checker.nullness.qual.NonNull;
 final class GrpcBlobReadChannel extends BaseStorageReadChannel<Object> {
 
   private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+  private final ResponseContentLifecycleManager responseContentLifecycleManager;
   private final ReadObjectRequest request;
   private final boolean autoGzipDecompression;
 
   GrpcBlobReadChannel(
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
+      ResponseContentLifecycleManager responseContentLifecycleManager,
       ReadObjectRequest request,
       boolean autoGzipDecompression) {
     super(Conversions.grpc().blobInfo());
     this.read = read;
+    this.responseContentLifecycleManager = responseContentLifecycleManager;
     this.request = request;
     this.autoGzipDecompression = autoGzipDecompression;
   }
@@ -53,7 +56,7 @@ final class GrpcBlobReadChannel extends BaseStorageReadChannel<Object> {
           ReadableByteChannelSessionBuilder b =
               ResumableMedia.gapic()
                   .read()
-                  .byteChannel(read)
+                  .byteChannel(read, responseContentLifecycleManager)
                   .setHasher(Hasher.noop())
                   .setAutoGzipDecompression(autoGzipDecompression);
           BufferHandle bufferHandle = getBufferHandle();
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 e9857e93..e5e81c06 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
@@ -180,6 +180,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
                   .collect(ImmutableSet.toImmutableSet())));
 
   final StorageClient storageClient;
+  final ResponseContentLifecycleManager responseContentLifecycleManager;
   final WriterFactory writerFactory;
   final GrpcConversions codecs;
   final GrpcRetryAlgorithmManager retryAlgorithmManager;
@@ -192,10 +193,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   GrpcStorageImpl(
       GrpcStorageOptions options,
       StorageClient storageClient,
+      ResponseContentLifecycleManager responseContentLifecycleManager,
       WriterFactory writerFactory,
       Opts<UserProject> defaultOpts) {
     super(options);
     this.storageClient = storageClient;
+    this.responseContentLifecycleManager = responseContentLifecycleManager;
     this.writerFactory = writerFactory;
     this.defaultOpts = defaultOpts;
     this.codecs = Conversions.grpc();
@@ -716,8 +719,10 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
     GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
+
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
+        responseContentLifecycleManager,
         request,
         !opts.autoGzipDecompression());
   }
@@ -1868,7 +1873,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
         opts.grpcMetadataMapper().apply(Retrying.newCallContext().withRetryableCodes(codes));
     return ResumableMedia.gapic()
         .read()
-        .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
+        .byteChannel(
+            storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
+            responseContentLifecycleManager)
         .setAutoGzipDecompression(!opts.autoGzipDecompression())
         .unbuffered()
         .setReadObjectRequest(readObjectRequest)
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 841343d3..438b5fd5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -25,14 +25,20 @@ import com.google.api.core.InternalApi;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.core.NoCredentialsProvider;
+import com.google.api.gax.grpc.GrpcCallSettings;
 import com.google.api.gax.grpc.GrpcInterceptorProvider;
+import com.google.api.gax.grpc.GrpcStubCallableFactory;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.StreamResumptionStrategy;
+import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.NoHeaderProvider;
+import com.google.api.gax.rpc.RequestParamsBuilder;
+import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.api.gax.rpc.internal.QuotaProjectIdHidingCredentials;
+import com.google.api.pathtemplate.PathTemplate;
 import com.google.auth.Credentials;
 import com.google.cloud.NoCredentials;
 import com.google.cloud.ServiceFactory;
@@ -47,25 +53,50 @@ import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.UnifiedOpts.UserProject;
 import com.google.cloud.storage.spi.StorageRpcFactory;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+import com.google.protobuf.UnsafeByteOperations;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.ReadObjectResponse;
 import com.google.storage.v2.StorageClient;
 import com.google.storage.v2.StorageSettings;
+import com.google.storage.v2.stub.GrpcStorageCallableFactory;
+import com.google.storage.v2.stub.GrpcStorageStub;
+import com.google.storage.v2.stub.StorageStub;
+import com.google.storage.v2.stub.StorageStubSettings;
 import io.grpc.ClientInterceptor;
+import io.grpc.Detachable;
+import io.grpc.HasByteBuffer;
+import io.grpc.KnownLength;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import io.grpc.protobuf.ProtoUtils;
+import java.io.Closeable;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.Serializable;
 import java.net.URI;
+import java.nio.ByteBuffer;
 import java.time.Clock;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Set;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -658,11 +689,32 @@ public final class GrpcStorageOptions extends StorageOptions
           Tuple<StorageSettings, Opts<UserProject>> t = grpcStorageOptions.resolveSettingsAndOpts();
           StorageSettings storageSettings = t.x();
           Opts<UserProject> defaultOpts = t.y();
-          return new GrpcStorageImpl(
-              grpcStorageOptions,
-              StorageClient.create(storageSettings),
-              grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
-              defaultOpts);
+          if (ZeroCopyReadinessChecker.isReady()) {
+            StorageStubSettings stubSettings =
+                (StorageStubSettings) storageSettings.getStubSettings();
+            ClientContext clientContext = ClientContext.create(stubSettings);
+            GrpcStorageCallableFactory grpcStorageCallableFactory =
+                new GrpcStorageCallableFactory();
+            InternalZeroCopyGrpcStorageStub stub =
+                new InternalZeroCopyGrpcStorageStub(
+                    stubSettings, clientContext, grpcStorageCallableFactory);
+            StorageClient client = new InternalStorageClient(stub);
+            return new GrpcStorageImpl(
+                grpcStorageOptions,
+                client,
+                stub.getObjectMediaResponseMarshaller,
+                grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
+                defaultOpts);
+          } else {
+            StorageClient client = StorageClient.create(storageSettings);
+            return new GrpcStorageImpl(
+                grpcStorageOptions,
+                client,
+                ResponseContentLifecycleManager.noop(),
+                grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
+                defaultOpts);
+          }
+
         } catch (IOException e) {
           throw new IllegalStateException(
               "Unable to instantiate gRPC com.google.cloud.storage.Storage client.", e);
@@ -780,4 +832,241 @@ public final class GrpcStorageOptions extends StorageOptions
       return INSTANCE;
     }
   }
+
+  private static final class InternalStorageClient extends StorageClient {
+
+    private InternalStorageClient(StorageStub stub) {
+      super(stub);
+    }
+  }
+
+  private static final class InternalZeroCopyGrpcStorageStub extends GrpcStorageStub
+      implements AutoCloseable {
+    private final ReadObjectResponseZeroCopyMessageMarshaller getObjectMediaResponseMarshaller;
+
+    private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse>
+        serverStreamingCallable;
+
+    private InternalZeroCopyGrpcStorageStub(
+        StorageStubSettings settings,
+        ClientContext clientContext,
+        GrpcStubCallableFactory callableFactory)
+        throws IOException {
+      super(settings, clientContext, callableFactory);
+
+      this.getObjectMediaResponseMarshaller =
+          new ReadObjectResponseZeroCopyMessageMarshaller(ReadObjectResponse.getDefaultInstance());
+
+      MethodDescriptor<ReadObjectRequest, ReadObjectResponse> readObjectMethodDescriptor =
+          MethodDescriptor.<ReadObjectRequest, ReadObjectResponse>newBuilder()
+              .setType(MethodDescriptor.MethodType.SERVER_STREAMING)
+              .setFullMethodName("google.storage.v2.Storage/ReadObject")
+              .setRequestMarshaller(ProtoUtils.marshaller(ReadObjectRequest.getDefaultInstance()))
+              .setResponseMarshaller(getObjectMediaResponseMarshaller)
+              .build();
+
+      GrpcCallSettings<ReadObjectRequest, ReadObjectResponse> readObjectTransportSettings =
+          GrpcCallSettings.<ReadObjectRequest, ReadObjectResponse>newBuilder()
+              .setMethodDescriptor(readObjectMethodDescriptor)
+              .setParamsExtractor(
+                  request -> {
+                    RequestParamsBuilder builder = RequestParamsBuilder.create();
+                    // todo: this is fragile to proto annotation changes, and would require manual
+                    // maintenance
+                    builder.add(request.getBucket(), "bucket", PathTemplate.create("{bucket=**}"));
+                    return builder.build();
+                  })
+              .build();
+
+      this.serverStreamingCallable =
+          callableFactory.createServerStreamingCallable(
+              readObjectTransportSettings, settings.readObjectSettings(), clientContext);
+    }
+
+    @Override
+    public ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> readObjectCallable() {
+      return serverStreamingCallable;
+    }
+  }
+
+  @VisibleForTesting
+  static class ReadObjectResponseZeroCopyMessageMarshaller
+      implements MethodDescriptor.PrototypeMarshaller<ReadObjectResponse>,
+          ResponseContentLifecycleManager,
+          Closeable {
+    private final Map<ReadObjectResponse, InputStream> unclosedStreams;
+    private final Parser<ReadObjectResponse> parser;
+    private final MethodDescriptor.PrototypeMarshaller<ReadObjectResponse> baseMarshaller;
+
+    ReadObjectResponseZeroCopyMessageMarshaller(ReadObjectResponse defaultInstance) {
+      parser = defaultInstance.getParserForType();
+      baseMarshaller =
+          (MethodDescriptor.PrototypeMarshaller<ReadObjectResponse>)
+              ProtoUtils.marshaller(defaultInstance);
+      unclosedStreams = Collections.synchronizedMap(new IdentityHashMap<>());
+    }
+
+    @Override
+    public Class<ReadObjectResponse> getMessageClass() {
+      return baseMarshaller.getMessageClass();
+    }
+
+    @Override
+    public ReadObjectResponse getMessagePrototype() {
+      return baseMarshaller.getMessagePrototype();
+    }
+
+    @Override
+    public InputStream stream(ReadObjectResponse value) {
+      return baseMarshaller.stream(value);
+    }
+
+    @Override
+    public ReadObjectResponse parse(InputStream stream) {
+      CodedInputStream cis = null;
+      try {
+        if (stream instanceof KnownLength
+            && stream instanceof Detachable
+            && stream instanceof HasByteBuffer
+            && ((HasByteBuffer) stream).byteBufferSupported()) {
+          int size = stream.available();
+          // Stream is now detached here and should be closed later.
+          stream = ((Detachable) stream).detach();
+          // This mark call is to keep buffer while traversing buffers using skip.
+          stream.mark(size);
+          List<ByteString> byteStrings = new ArrayList<>();
+          while (stream.available() != 0) {
+            ByteBuffer buffer = ((HasByteBuffer) stream).getByteBuffer();
+            byteStrings.add(UnsafeByteOperations.unsafeWrap(buffer));
+            stream.skip(buffer.remaining());
+          }
+          stream.reset();
+          cis = ByteString.copyFrom(byteStrings).newCodedInput();
+          cis.enableAliasing(true);
+          cis.setSizeLimit(Integer.MAX_VALUE);
+        }
+      } catch (IOException e) {
+        throw Status.INTERNAL
+            .withDescription("Error parsing input stream for ReadObject")
+            .withCause(e)
+            .asRuntimeException();
+      }
+      if (cis != null) {
+        // fast path (no memory copy)
+        ReadObjectResponse message;
+        try {
+          message = parseFrom(cis);
+        } catch (InvalidProtocolBufferException ipbe) {
+          throw Status.INTERNAL
+              .withDescription("Invalid protobuf byte sequence for ReadObject")
+              .withCause(ipbe)
+              .asRuntimeException();
+        }
+        unclosedStreams.put(message, stream);
+        return message;
+      } else {
+        // slow path
+        return baseMarshaller.parse(stream);
+      }
+    }
+
+    private ReadObjectResponse parseFrom(CodedInputStream stream)
+        throws InvalidProtocolBufferException {
+      ReadObjectResponse message = parser.parseFrom(stream);
+      try {
+        stream.checkLastTagWas(0);
+        return message;
+      } catch (InvalidProtocolBufferException e) {
+        e.setUnfinishedMessage(message);
+        throw e;
+      }
+    }
+
+    @Override
+    public ResponseContentLifecycleHandle get(ReadObjectResponse response) {
+      InputStream stream = unclosedStreams.remove(response);
+      return new ResponseContentLifecycleHandle(response, stream);
+    }
+
+    @Override
+    public void close() throws IOException {
+      closeAllStreams(unclosedStreams.values());
+    }
+
+    /**
+     * In the event closing the streams results in multiple streams throwing IOExceptions, collect
+     * them all as suppressed exceptions on the first occurrence.
+     */
+    @VisibleForTesting
+    static void closeAllStreams(Collection<InputStream> inputStreams) throws IOException {
+      IOException ioException =
+          inputStreams.stream()
+              .map(
+                  stream -> {
+                    try {
+                      stream.close();
+                      return null;
+                    } catch (IOException e) {
+                      return e;
+                    }
+                  })
+              .filter(Objects::nonNull)
+              .reduce(
+                  null,
+                  (l, r) -> {
+                    if (l != null) {
+                      l.addSuppressed(r);
+                      return l;
+                    } else {
+                      return r;
+                    }
+                  },
+                  (l, r) -> l);
+
+      if (ioException != null) {
+        throw ioException;
+      }
+    }
+  }
+
+  static final class ZeroCopyReadinessChecker {
+    private static final boolean isZeroCopyReady;
+
+    static {
+      // Check whether io.grpc.Detachable exists?
+      boolean detachableClassExists = false;
+      try {
+        // Try to load Detachable interface in the package where KnownLength is in.
+        // This can be done directly by looking up io.grpc.Detachable but rather
+        // done indirectly to handle the case where gRPC is being shaded in a
+        // different package.
+        String knownLengthClassName = KnownLength.class.getName();
+        String detachableClassName =
+            knownLengthClassName.substring(0, knownLengthClassName.lastIndexOf('.') + 1)
+                + "Detachable";
+        Class<?> detachableClass = Class.forName(detachableClassName);
+        detachableClassExists = (detachableClass != null);
+      } catch (ClassNotFoundException ex) {
+        // leaves detachableClassExists false
+      }
+      // Check whether com.google.protobuf.UnsafeByteOperations exists?
+      boolean unsafeByteOperationsClassExists = false;
+      try {
+        // Same above
+        String messageLiteClassName = MessageLite.class.getName();
+        String unsafeByteOperationsClassName =
+            messageLiteClassName.substring(0, messageLiteClassName.lastIndexOf('.') + 1)
+                + "UnsafeByteOperations";
+        Class<?> unsafeByteOperationsClass = Class.forName(unsafeByteOperationsClassName);
+        unsafeByteOperationsClassExists = (unsafeByteOperationsClass != null);
+      } catch (ClassNotFoundException ex) {
+        // leaves unsafeByteOperationsClassExists false
+      }
+      isZeroCopyReady = detachableClassExists && unsafeByteOperationsClassExists;
+    }
+
+    public static boolean isReady() {
+      return isZeroCopyReady;
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
index 06fca041..d7741b7d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Hasher.java
@@ -20,6 +20,7 @@ import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.common.hash.Hashing;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.function.Supplier;
 import javax.annotation.concurrent.Immutable;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -36,6 +37,8 @@ interface Hasher {
 
   void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException;
 
+  void validate(Crc32cValue<?> expected, List<ByteBuffer> buffers) throws IOException;
+
   @Nullable
   Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2);
 
@@ -61,6 +64,9 @@ interface Hasher {
     @Override
     public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) {}
 
+    @Override
+    public void validate(Crc32cValue<?> expected, List<ByteBuffer> b) {}
+
     @Override
     public @Nullable Crc32cLengthKnown nullSafeConcat(Crc32cLengthKnown r1, Crc32cLengthKnown r2) {
       return null;
@@ -79,7 +85,24 @@ interface Hasher {
       return Crc32cValue.of(Hashing.crc32c().hashBytes(b).asInt(), remaining);
     }
 
-    @SuppressWarnings("ConstantConditions")
+    @SuppressWarnings({"ConstantConditions", "UnstableApiUsage"})
+    @Override
+    public void validate(Crc32cValue<?> expected, List<ByteBuffer> b) throws IOException {
+      long remaining = 0;
+      com.google.common.hash.Hasher crc32c = Hashing.crc32c().newHasher();
+      for (ByteBuffer tmp : b) {
+        remaining += tmp.remaining();
+        crc32c.putBytes(tmp);
+      }
+      Crc32cLengthKnown actual = Crc32cValue.of(crc32c.hash().asInt(), remaining);
+      if (!actual.eqValue(expected)) {
+        throw new IOException(
+            String.format(
+                "Mismatch checksum value. Expected %s actual %s",
+                expected.debugString(), actual.debugString()));
+      }
+    }
+
     @Override
     public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b) throws IOException {
       Crc32cLengthKnown actual = hash(b);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
index 192cfec6..622988d4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java
@@ -120,7 +120,7 @@ final class JsonResumableSessionPutTask
         }
       } else if (finalizing && JsonResumableSessionFailureScenario.isOk(code)) {
         @Nullable StorageObject storageObject;
-        @Nullable BigInteger actualSize;
+        BigInteger actualSize = BigInteger.ZERO;
 
         Long contentLength = response.getHeaders().getContentLength();
         String contentType = response.getHeaders().getContentType();
@@ -130,7 +130,12 @@ final class JsonResumableSessionPutTask
         boolean isJson = contentType != null && contentType.startsWith("application/json");
         if (isJson) {
           storageObject = response.parseAs(StorageObject.class);
-          actualSize = storageObject != null ? storageObject.getSize() : null;
+          if (storageObject != null) {
+            BigInteger size = storageObject.getSize();
+            if (size != null) {
+              actualSize = size;
+            }
+          }
         } else if ((contentLength == null || contentLength == 0) && storedContentLength != null) {
           // when a signed url is used, the finalize response is empty
           response.ignore();
@@ -150,7 +155,6 @@ final class JsonResumableSessionPutTask
         int compare = expectedSize.compareTo(actualSize);
         if (compare == 0) {
           success = true;
-          //noinspection DataFlowIssue  compareTo result will filter out actualSize == null
           return ResumableOperationResult.complete(storageObject, actualSize.longValue());
         } else if (compare > 0) {
           StorageException se =
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadCursor.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadCursor.java
new file mode 100644
index 00000000..b65a6e25
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ReadCursor.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.storage;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+/**
+ * Shrink wraps a beginning, offset and limit for tracking state of an individual invocation of
+ * {@link #read}
+ */
+final class ReadCursor {
+  private final long beginning;
+  private long offset;
+  private final long limit;
+
+  ReadCursor(long beginning, long limit) {
+    this.limit = limit;
+    this.beginning = beginning;
+    this.offset = beginning;
+  }
+
+  public boolean hasRemaining() {
+    return limit - offset > 0;
+  }
+
+  public void advance(long incr) {
+    checkArgument(incr >= 0);
+    offset += incr;
+  }
+
+  public long read() {
+    return offset - beginning;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("ReadCursor{begin=%d, offset=%d, limit=%d}", beginning, offset, limit);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleHandle.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleHandle.java
new file mode 100644
index 00000000..20fc3658
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleHandle.java
@@ -0,0 +1,57 @@
+/*
+ * 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.storage.v2.ReadObjectResponse;
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class ResponseContentLifecycleHandle implements Closeable {
+  @Nullable private final Closeable dispose;
+
+  private final List<ByteBuffer> buffers;
+
+  ResponseContentLifecycleHandle(ReadObjectResponse resp, @Nullable Closeable dispose) {
+    this.dispose = dispose;
+
+    this.buffers = resp.getChecksummedData().getContent().asReadOnlyByteBufferList();
+  }
+
+  void copy(ReadCursor c, ByteBuffer[] dsts, int offset, int length) {
+    for (ByteBuffer b : buffers) {
+      long copiedBytes = Buffers.copy(b, dsts, offset, length);
+      c.advance(copiedBytes);
+      if (b.hasRemaining()) break;
+    }
+  }
+
+  boolean hasRemaining() {
+    for (ByteBuffer b : buffers) {
+      if (b.hasRemaining()) return true;
+    }
+    return false;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (dispose != null) {
+      dispose.close();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleManager.java
new file mode 100644
index 00000000..32365133
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleManager.java
@@ -0,0 +1,31 @@
+/*
+ * 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.storage.v2.ReadObjectResponse;
+
+interface ResponseContentLifecycleManager {
+  ResponseContentLifecycleHandle get(ReadObjectResponse response);
+
+  static ResponseContentLifecycleManager noop() {
+    return response ->
+        new ResponseContentLifecycleHandle(
+            response,
+            () -> {
+              // no-op
+            });
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index f8b4f94a..eeb24a44 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -118,6 +118,9 @@ public final class StorageException extends BaseHttpServiceException {
     if (t instanceof ApiException) {
       return asStorageException((ApiException) t);
     }
+    if (t.getCause() instanceof ApiException) {
+      return asStorageException((ApiException) t.getCause());
+    }
     return getStorageException(t);
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
index 53ad6142..ab32532a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
@@ -22,14 +22,17 @@ import com.google.cloud.NoCredentials;
 import com.google.cloud.ServiceDefaults;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.http.HttpTransportOptions;
+import com.google.cloud.storage.GrpcStorageOptions.GrpcStorageDefaults;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageDefaults;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageFactory;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageRpcFactory;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 public abstract class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
 
@@ -95,6 +98,18 @@ public abstract class StorageOptions extends ServiceOptions<Storage, StorageOpti
 
     public abstract Builder setStorageRetryStrategy(StorageRetryStrategy storageRetryStrategy);
 
+    /**
+     * @see BlobWriteSessionConfig
+     * @see BlobWriteSessionConfigs
+     * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+     * @see HttpStorageDefaults#getDefaultStorageWriterConfig()
+     * @see GrpcStorageDefaults#getDefaultStorageWriterConfig()
+     * @since 2.37.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public abstract StorageOptions.Builder setBlobWriteSessionConfig(
+        @NonNull BlobWriteSessionConfig blobWriteSessionConfig);
+
     @Override
     public abstract StorageOptions build();
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java
index fa1ed0c9..e466a824 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java
@@ -19,9 +19,16 @@ package com.google.cloud.storage.transfermanager;
 final class DefaultQos implements Qos {
 
   private final long divideAndConquerThreshold;
+  private final long parallelCompositeUploadThreshold;
+  private boolean threadThresholdMet;
 
-  private DefaultQos(long divideAndConquerThreshold) {
+  private DefaultQos(
+      long divideAndConquerThreshold,
+      long parallelCompositeUploadThreshold,
+      boolean threadThresholdMet) {
     this.divideAndConquerThreshold = divideAndConquerThreshold;
+    this.parallelCompositeUploadThreshold = parallelCompositeUploadThreshold;
+    this.threadThresholdMet = threadThresholdMet;
   }
 
   @Override
@@ -29,11 +36,13 @@ final class DefaultQos implements Qos {
     return objectSize > divideAndConquerThreshold;
   }
 
-  static DefaultQos of() {
-    return of(128L * 1024 * 1024);
+  @Override
+  public boolean parallelCompositeUpload(long objectSize) {
+    return threadThresholdMet && objectSize > parallelCompositeUploadThreshold;
   }
 
-  static DefaultQos of(long divideAndConquerThreshold) {
-    return new DefaultQos(divideAndConquerThreshold);
+  static DefaultQos of(TransferManagerConfig config) {
+    return new DefaultQos(
+        128L * 1024 * 1024, 4L * config.getPerWorkerBufferSize(), config.getMaxWorkers() > 2);
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelCompositeUploadCallable.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelCompositeUploadCallable.java
new file mode 100644
index 00000000..d306c4a6
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelCompositeUploadCallable.java
@@ -0,0 +1,94 @@
+/*
+ * 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.transfermanager;
+
+import com.google.api.core.ApiFuture;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+import com.google.common.io.ByteStreams;
+import java.nio.channels.FileChannel;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+final class ParallelCompositeUploadCallable implements Callable<UploadResult> {
+  private final Storage storage;
+
+  private final BlobInfo originalBlob;
+
+  private final Path sourceFile;
+
+  private final ParallelUploadConfig parallelUploadConfig;
+
+  private final Storage.BlobWriteOption[] opts;
+
+  public ParallelCompositeUploadCallable(
+      Storage storage,
+      BlobInfo originalBlob,
+      Path sourceFile,
+      ParallelUploadConfig parallelUploadConfig,
+      BlobWriteOption[] opts) {
+    this.storage = storage;
+    this.originalBlob = originalBlob;
+    this.sourceFile = sourceFile;
+    this.parallelUploadConfig = parallelUploadConfig;
+    this.opts = opts;
+  }
+
+  public UploadResult call() {
+    return uploadPCU();
+  }
+
+  private UploadResult uploadPCU() {
+    BlobWriteSession session = storage.blobWriteSession(originalBlob, opts);
+    try (WritableByteChannel writableByteChannel = session.open();
+        FileChannel fc = FileChannel.open(sourceFile, StandardOpenOption.READ)) {
+      ByteStreams.copy(fc, writableByteChannel);
+    } catch (StorageException e) {
+      if (parallelUploadConfig.isSkipIfExists() && e.getCode() == 412) {
+        return UploadResult.newBuilder(originalBlob, TransferStatus.SKIPPED)
+            .setException(e)
+            .build();
+      } else {
+        return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+            .setException(e)
+            .build();
+      }
+    } catch (Exception e) {
+      return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+    try {
+      ApiFuture<BlobInfo> result = session.getResult();
+      BlobInfo newBlob = result.get(10, TimeUnit.SECONDS);
+      return UploadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
+          .setUploadedBlob(newBlob)
+          .build();
+    } catch (InterruptedException | ExecutionException | TimeoutException e) {
+      return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java
index 10059164..492cf3d1 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/Qos.java
@@ -19,4 +19,6 @@ package com.google.cloud.storage.transfermanager;
 interface Qos {
 
   boolean divideAndConquer(long objectSize);
+
+  boolean parallelCompositeUpload(long objectSize);
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
index f333deca..9aae2345 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManager.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage.transfermanager;
 
 import com.google.api.core.BetaApi;
 import com.google.cloud.storage.BlobInfo;
+import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -58,7 +59,7 @@ public interface TransferManager extends AutoCloseable {
    */
   @BetaApi
   @NonNull
-  UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config);
+  UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config) throws IOException;
 
   /**
    * Downloads a list of blobs in parallel. This operation will not block the invoking thread,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
index b638e46f..f06839dd 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerConfig.java
@@ -34,21 +34,18 @@ public final class TransferManagerConfig {
   private final boolean allowParallelCompositeUpload;
 
   private final StorageOptions storageOptions;
-  private final Qos qos;
 
   TransferManagerConfig(
       int maxWorkers,
       int perWorkerBufferSize,
       boolean allowDivideAndConquerDownload,
       boolean allowParallelCompositeUpload,
-      StorageOptions storageOptions,
-      Qos qos) {
+      StorageOptions storageOptions) {
     this.maxWorkers = maxWorkers;
     this.perWorkerBufferSize = perWorkerBufferSize;
     this.allowDivideAndConquerDownload = allowDivideAndConquerDownload;
     this.allowParallelCompositeUpload = allowParallelCompositeUpload;
     this.storageOptions = storageOptions;
-    this.qos = qos;
   }
 
   /**
@@ -86,6 +83,15 @@ public final class TransferManagerConfig {
    * chunking will be beneficial
    *
    * @see Builder#setAllowParallelCompositeUpload(boolean)
+   *     <p>Note: Performing parallel composite uploads costs more money. <a
+   *     href="https://cloud.google.com/storage/pricing#operations-by-class">Class A</a> operations
+   *     are performed to create each part and to perform each compose. If a storage tier other than
+   *     <a href="https://cloud.google.com/storage/docs/storage-classes"><code>STANDARD</code></a>
+   *     is used, early deletion fees apply to deletion of the parts.
+   *     <p>Please see the <a
+   *     href="https://cloud.google.com/storage/docs/parallel-composite-uploads">Parallel composite
+   *     uploads</a> documentation for a more in depth explanation of the limitations of Parallel
+   *     composite uploads.
    */
   @BetaApi
   public boolean isAllowParallelCompositeUpload() {
@@ -105,7 +111,7 @@ public final class TransferManagerConfig {
   /** The service object for {@link TransferManager} */
   @BetaApi
   public TransferManager getService() {
-    return new TransferManagerImpl(this);
+    return new TransferManagerImpl(this, DefaultQos.of(this));
   }
 
   @BetaApi
@@ -115,14 +121,9 @@ public final class TransferManagerConfig {
         .setAllowParallelCompositeUpload(allowParallelCompositeUpload)
         .setMaxWorkers(maxWorkers)
         .setPerWorkerBufferSize(perWorkerBufferSize)
-        .setQos(qos)
         .setStorageOptions(storageOptions);
   }
 
-  Qos getQos() {
-    return qos;
-  }
-
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -179,7 +180,6 @@ public final class TransferManagerConfig {
     private boolean allowParallelCompositeUpload;
 
     private StorageOptions storageOptions;
-    private Qos qos;
 
     private Builder() {
       this.perWorkerBufferSize = 16 * 1024 * 1024;
@@ -187,7 +187,6 @@ public final class TransferManagerConfig {
       this.allowDivideAndConquerDownload = false;
       this.allowParallelCompositeUpload = false;
       this.storageOptions = StorageOptions.getDefaultInstance();
-      this.qos = DefaultQos.of();
     }
 
     /**
@@ -244,8 +243,8 @@ public final class TransferManagerConfig {
      * @see TransferManagerConfig#isAllowDivideAndConquerDownload()
      */
     @BetaApi
-    public Builder setAllowParallelCompositeUpload(boolean allowDivideAndConquerDownload) {
-      this.allowDivideAndConquerDownload = allowDivideAndConquerDownload;
+    public Builder setAllowParallelCompositeUpload(boolean allowParallelCompositeUpload) {
+      this.allowParallelCompositeUpload = allowParallelCompositeUpload;
       return this;
     }
 
@@ -263,12 +262,6 @@ public final class TransferManagerConfig {
       return this;
     }
 
-    @BetaApi
-    Builder setQos(Qos qos) {
-      this.qos = qos;
-      return this;
-    }
-
     /**
      * Creates a TransferManagerConfig object.
      *
@@ -281,8 +274,7 @@ public final class TransferManagerConfig {
           perWorkerBufferSize,
           allowDivideAndConquerDownload,
           allowParallelCompositeUpload,
-          storageOptions,
-          qos);
+          storageOptions);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
index b731c2c3..4bbbeb93 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java
@@ -20,9 +20,13 @@ import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.ListenableFutureToApiFuture;
+import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageOptions;
@@ -31,10 +35,13 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Deque;
 import java.util.List;
+import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BinaryOperator;
@@ -50,12 +57,17 @@ final class TransferManagerImpl implements TransferManager {
   private final Qos qos;
   private final Storage storage;
 
-  TransferManagerImpl(TransferManagerConfig transferManagerConfig) {
+  private final Deque<PendingPcuTask> pcuQueue;
+  // define a unique object which we can use to synchronize modification of pcuPoller
+  private final Object pcuPollerSync = new Object();
+  private volatile ApiFuture<?> pcuPoller;
+
+  TransferManagerImpl(TransferManagerConfig transferManagerConfig, Qos qos) {
     this.transferManagerConfig = transferManagerConfig;
     this.executor =
         MoreExecutors.listeningDecorator(
             Executors.newFixedThreadPool(transferManagerConfig.getMaxWorkers()));
-    this.qos = transferManagerConfig.getQos();
+    this.qos = qos;
     StorageOptions storageOptions = transferManagerConfig.getStorageOptions();
     String userAgent = storageOptions.getUserAgent();
     if (userAgent == null || !userAgent.contains(USER_AGENT_ENTRY)) {
@@ -67,6 +79,14 @@ final class TransferManagerImpl implements TransferManager {
                       ImmutableMap.of("User-Agent", USER_AGENT_ENTRY + LIBRARY_VERSION)))
               .build();
     }
+    // Create the blobWriteSessionConfig for ParallelCompositeUpload
+    if (transferManagerConfig.isAllowParallelCompositeUpload()) {
+      ParallelCompositeUploadBlobWriteSessionConfig pcuConfig =
+          BlobWriteSessionConfigs.parallelCompositeUpload()
+              .withExecutorSupplier(ExecutorSupplier.useExecutor(executor));
+      storageOptions = storageOptions.toBuilder().setBlobWriteSessionConfig(pcuConfig).build();
+    }
+    this.pcuQueue = new ConcurrentLinkedDeque<>();
     this.storage = storageOptions.getService();
   }
 
@@ -79,7 +99,8 @@ final class TransferManagerImpl implements TransferManager {
 
   @Override
   @BetaApi
-  public @NonNull UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config) {
+  public @NonNull UploadJob uploadFiles(List<Path> files, ParallelUploadConfig config)
+      throws IOException {
     Storage.BlobWriteOption[] opts =
         config.getWriteOptsPerRequest().toArray(new BlobWriteOption[0]);
     List<ApiFuture<UploadResult>> uploadTasks = new ArrayList<>();
@@ -87,9 +108,19 @@ final class TransferManagerImpl implements TransferManager {
       if (Files.isDirectory(file)) throw new IllegalStateException("Directories are not supported");
       String blobName = TransferManagerUtils.createBlobName(config, file);
       BlobInfo blobInfo = BlobInfo.newBuilder(config.getBucketName(), blobName).build();
-      UploadCallable callable =
-          new UploadCallable(transferManagerConfig, storage, blobInfo, file, config, opts);
-      uploadTasks.add(convert(executor.submit(callable)));
+      if (transferManagerConfig.isAllowParallelCompositeUpload()
+          && qos.parallelCompositeUpload(Files.size(file))) {
+        ParallelCompositeUploadCallable callable =
+            new ParallelCompositeUploadCallable(storage, blobInfo, file, config, opts);
+        SettableApiFuture<UploadResult> resultFuture = SettableApiFuture.create();
+        pcuQueue.add(new PendingPcuTask(callable, resultFuture));
+        uploadTasks.add(resultFuture);
+        schedulePcuPoller();
+      } else {
+        UploadCallable callable =
+            new UploadCallable(transferManagerConfig, storage, blobInfo, file, config, opts);
+        uploadTasks.add(convert(executor.submit(callable)));
+      }
     }
     return UploadJob.newBuilder()
         .setParallelUploadConfig(config)
@@ -151,6 +182,26 @@ final class TransferManagerImpl implements TransferManager {
         .build();
   }
 
+  private void schedulePcuPoller() {
+    if (pcuPoller == null) {
+      synchronized (pcuPollerSync) {
+        if (pcuPoller == null) {
+          pcuPoller = convert(executor.submit(new PcuPoller()));
+        }
+      }
+    }
+  }
+
+  private void deschedulePcuPoller() {
+    if (pcuPoller != null) {
+      synchronized (pcuPollerSync) {
+        if (pcuPoller != null) {
+          pcuPoller = null;
+        }
+      }
+    }
+  }
+
   private static <T> ApiFuture<T> convert(ListenableFuture<T> lf) {
     return new ListenableFutureToApiFuture<>(lf);
   }
@@ -191,4 +242,43 @@ final class TransferManagerImpl implements TransferManager {
       return new Range(begin, end);
     }
   }
+
+  /**
+   * When performing a Parallel composite upload, the thread pool we perform work on is shared as
+   * the PCU worker pool. Because of this, if we submit our work to the executor service and take
+   * all the threads waiting for PCU uploads to complete, the PCU work doesn't have any threads
+   * available to itself.
+   *
+   * <p>This class represents a single worker that will be submitted to the executor service and
+   * will poll a queue to process a single PCU at a time, leaving any other threads free for PCU
+   * work.
+   */
+  private final class PcuPoller implements Runnable {
+
+    @Override
+    public void run() {
+      do {
+        PendingPcuTask poll = pcuQueue.poll();
+        if (poll == null) {
+          deschedulePcuPoller();
+          return;
+        }
+
+        UploadResult result = poll.callable.call();
+        poll.resultFuture.set(result);
+
+      } while (true);
+    }
+  }
+
+  private static final class PendingPcuTask {
+    private final ParallelCompositeUploadCallable callable;
+    private final SettableApiFuture<UploadResult> resultFuture;
+
+    private PendingPcuTask(
+        ParallelCompositeUploadCallable callable, SettableApiFuture<UploadResult> resultFuture) {
+      this.callable = callable;
+      this.resultFuture = resultFuture;
+    }
+  }
 }
diff --git a/samples/snippets/src/main/java/com/example/storage/object/UploadObjectFromMemory.java b/samples/snippets/src/main/java/com/example/storage/object/UploadObjectFromMemory.java
index e0190ee8..98d00202 100644
--- a/samples/snippets/src/main/java/com/example/storage/object/UploadObjectFromMemory.java
+++ b/samples/snippets/src/main/java/com/example/storage/object/UploadObjectFromMemory.java
@@ -22,7 +22,6 @@ import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 
@@ -45,7 +44,25 @@ public class UploadObjectFromMemory {
     BlobId blobId = BlobId.of(bucketName, objectName);
     BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
     byte[] content = contents.getBytes(StandardCharsets.UTF_8);
-    storage.createFrom(blobInfo, new ByteArrayInputStream(content));
+
+    // Optional: set a generation-match precondition to enable automatic retries, avoid potential
+    // race
+    // conditions and data corruptions. The request returns a 412 error if the
+    // preconditions are not met.
+    Storage.BlobTargetOption precondition;
+    if (storage.get(bucketName, objectName) == null) {
+      // For a target object that does not yet exist, set the DoesNotExist precondition.
+      // This will cause the request to fail if the object is created before the request runs.
+      precondition = Storage.BlobTargetOption.doesNotExist();
+    } else {
+      // If the destination already exists in your bucket, instead set a generation-match
+      // precondition. This will cause the request to fail if the existing object's generation
+      // changes before the request runs.
+      precondition =
+          Storage.BlobTargetOption.generationMatch(
+              storage.get(bucketName, objectName).getGeneration());
+    }
+    storage.create(blobInfo, content, precondition);
 
     System.out.println(
         "Object "
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadBucket.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadBucket.java
new file mode 100644
index 00000000..5e09f5a9
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadBucket.java
@@ -0,0 +1,60 @@
+/*
+ * 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_bucket]
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+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;
+import java.util.stream.Collectors;
+
+class DownloadBucket {
+
+  public static void downloadBucketContents(String projectId,
+      String bucketName, Path destinationDirectory) {
+    Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+    List<BlobInfo> blobs = storage
+        .list(bucketName)
+        .streamAll()
+        .map(blob -> blob.asBlobInfo())
+        .collect(Collectors.toList());
+    TransferManager transferManager = TransferManagerConfig.newBuilder().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_bucket]
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadMany.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadMany.java
new file mode 100644
index 00000000..4fd1720b
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadMany.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_download_many]
+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 DownloadMany {
+
+  public static void downloadManyBlobs(String bucketName,
+      List<BlobInfo> blobs, Path destinationDirectory) {
+
+    TransferManager transferManager = TransferManagerConfig.newBuilder().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_many]
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadDirectory.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadDirectory.java
new file mode 100644
index 00000000..096b2dae
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadDirectory.java
@@ -0,0 +1,59 @@
+/*
+ * 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_directory]
+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.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+class UploadDirectory {
+
+  public static void uploadDirectoryContents(String bucketName, Path sourceDirectory)
+      throws IOException {
+    TransferManager transferManager = TransferManagerConfig.newBuilder().build().getService();
+    ParallelUploadConfig parallelUploadConfig = ParallelUploadConfig
+        .newBuilder()
+        .setBucketName(bucketName)
+        .build();
+
+    // Create a list to store the file paths
+    List<Path> filePaths = new ArrayList<>();
+    // Get all files in the directory
+    // try-with-resource to ensure pathStream is closed
+    try (Stream<Path> pathStream = Files.walk(sourceDirectory)) {
+      pathStream.filter(Files::isRegularFile)
+          .forEach(filePaths::add);
+    }
+    List<UploadResult> results = transferManager
+        .uploadFiles(filePaths, parallelUploadConfig)
+        .getUploadResults();
+    for (UploadResult result : results) {
+      System.out.println("Upload for " + result.getInput().getName()
+          + " completed with status " + result.getStatus());
+    }
+  }
+
+}
+// [END storage_transfer_manager_upload_directory]
\ No newline at end of file
diff --git a/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadMany.java b/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadMany.java
new file mode 100644
index 00000000..245df364
--- /dev/null
+++ b/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadMany.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example.storage.transfermanager;
+
+// [START storage_transfer_manager_upload_many]
+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 UploadMany {
+
+  public static void uploadManyFiles(String bucketName, List<Path> files)
+      throws IOException {
+    TransferManager transferManager = TransferManagerConfig.newBuilder().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_many]


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
index ba9a972c..3a4fd7e9 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java
@@ -127,7 +127,8 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
                           .withDefaultCallContext(
                               contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
                       start,
-                      Hasher.noop()));
+                      Hasher.noop(),
+                      ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
         c.read(ByteBuffer.wrap(actualBytes));
@@ -155,7 +156,8 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
                           .withDefaultCallContext(
                               contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
                       start,
-                      Hasher.noop()));
+                      Hasher.noop(),
+                      ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(actualBytes, 2);
       try (UnbufferedReadableByteChannel c = session.open()) {
@@ -213,7 +215,8 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
                           .withDefaultCallContext(
                               contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
                       start,
-                      Hasher.noop()));
+                      Hasher.noop(),
+                      ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
         IOException ioException =
@@ -260,7 +263,8 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
                           .withDefaultCallContext(
                               contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
                       start,
-                      Hasher.enabled()));
+                      Hasher.enabled(),
+                      ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[40];
       try (UnbufferedReadableByteChannel c = session.open()) {
         IOException ioException =
@@ -299,7 +303,8 @@ public final class ITGapicUnbufferedReadableByteChannelTest {
                           .withDefaultCallContext(
                               contextWithRetryForCodes(StatusCode.Code.DATA_LOSS)),
                       start,
-                      Hasher.enabled()));
+                      Hasher.enabled(),
+                      ResponseContentLifecycleManager.noop()));
       byte[] actualBytes = new byte[41];
       //noinspection resource
       UnbufferedReadableByteChannel c = session.open();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
index 2e8efa05..41f6df1d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGzipReadableByteChannelTest.java
@@ -124,7 +124,9 @@ public class ITGzipReadableByteChannelTest {
       UnbufferedReadableByteChannelSession<Object> session =
           ResumableMedia.gapic()
               .read()
-              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .byteChannel(
+                  storageClient.getInstance().readObjectCallable(),
+                  ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(true)
               .unbuffered()
@@ -143,7 +145,9 @@ public class ITGzipReadableByteChannelTest {
       UnbufferedReadableByteChannelSession<Object> session =
           ResumableMedia.gapic()
               .read()
-              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .byteChannel(
+                  storageClient.getInstance().readObjectCallable(),
+                  ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(false)
               .unbuffered()
@@ -193,7 +197,9 @@ public class ITGzipReadableByteChannelTest {
       UnbufferedReadableByteChannelSession<Object> session =
           ResumableMedia.gapic()
               .read()
-              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .byteChannel(
+                  storageClient.getInstance().readObjectCallable(),
+                  ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(true)
               .unbuffered()
@@ -212,7 +218,9 @@ public class ITGzipReadableByteChannelTest {
       UnbufferedReadableByteChannelSession<Object> session =
           ResumableMedia.gapic()
               .read()
-              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .byteChannel(
+                  storageClient.getInstance().readObjectCallable(),
+                  ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .setAutoGzipDecompression(false)
               .unbuffered()
@@ -231,7 +239,9 @@ public class ITGzipReadableByteChannelTest {
       UnbufferedReadableByteChannelSession<Object> session =
           ResumableMedia.gapic()
               .read()
-              .byteChannel(storageClient.getInstance().readObjectCallable())
+              .byteChannel(
+                  storageClient.getInstance().readObjectCallable(),
+                  ResponseContentLifecycleManager.noop())
               .setHasher(Hasher.noop())
               .unbuffered()
               .setReadObjectRequest(reqCompressed)
@@ -314,7 +324,7 @@ public class ITGzipReadableByteChannelTest {
         ReadableByteChannelSession<?, Object> session =
             ResumableMedia.gapic()
                 .read()
-                .byteChannel(sc.readObjectCallable())
+                .byteChannel(sc.readObjectCallable(), ResponseContentLifecycleManager.noop())
                 .setHasher(Hasher.noop())
                 .setAutoGzipDecompression(true)
                 .unbuffered()
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
index 76219bec..32d308a3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
@@ -38,7 +38,9 @@ public final class TransportCompatibilityTest {
             .setCredentials(NoCredentials.getInstance())
             .build();
     @SuppressWarnings("resource")
-    Storage s = new GrpcStorageImpl(options, null, null, Opts.empty());
+    Storage s =
+        new GrpcStorageImpl(
+            options, null, ResponseContentLifecycleManager.noop(), null, Opts.empty());
     ImmutableList<String> messages =
         Stream.<Supplier<?>>of(
                 s::batch,
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ZeroCopyMarshallerTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ZeroCopyMarshallerTest.java
new file mode 100644
index 00000000..57d96dfb
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ZeroCopyMarshallerTest.java
@@ -0,0 +1,223 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.TestUtils.getChecksummedData;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.storage.GrpcStorageOptions.ReadObjectResponseZeroCopyMessageMarshaller;
+import com.google.common.collect.ImmutableList;
+import com.google.common.hash.Hashing;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.ContentRange;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.ObjectChecksums;
+import com.google.storage.v2.ReadObjectResponse;
+import io.grpc.StatusRuntimeException;
+import io.grpc.internal.ReadableBuffer;
+import io.grpc.internal.ReadableBuffers;
+import java.io.Closeable;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import org.junit.Test;
+
+public class ZeroCopyMarshallerTest {
+  private final byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+  private final ByteString data = ByteString.copyFrom(bytes, 0, 10);
+  private final ReadObjectResponse response =
+      ReadObjectResponse.newBuilder()
+          .setMetadata(
+              Object.newBuilder()
+                  .setName("name")
+                  .setGeneration(3L)
+                  .setContentType("application/octet-stream")
+                  .build())
+          .setContentRange(ContentRange.newBuilder().setStart(0).build())
+          .setObjectChecksums(
+              ObjectChecksums.newBuilder().setCrc32C(Hashing.crc32c().hashBytes(bytes).asInt()))
+          .setChecksummedData(getChecksummedData(data, Hasher.enabled()))
+          .build();
+
+  private ReadObjectResponseZeroCopyMessageMarshaller createMarshaller() {
+    return new ReadObjectResponseZeroCopyMessageMarshaller(ReadObjectResponse.getDefaultInstance());
+  }
+
+  private byte[] dropLastOneByte(byte[] bytes) {
+    return Arrays.copyOfRange(bytes, 0, bytes.length - 1);
+  }
+
+  private InputStream createInputStream(byte[] bytes, boolean isZeroCopyable) {
+    ReadableBuffer buffer =
+        isZeroCopyable ? ReadableBuffers.wrap(ByteBuffer.wrap(bytes)) : ReadableBuffers.wrap(bytes);
+    return ReadableBuffers.openStream(buffer, true);
+  }
+
+  @Test
+  public void testParseOnFastPath() throws IOException {
+    InputStream stream = createInputStream(response.toByteArray(), true);
+    ReadObjectResponseZeroCopyMessageMarshaller marshaller = createMarshaller();
+    ReadObjectResponse response = marshaller.parse(stream);
+    assertEquals(response, this.response);
+    ResponseContentLifecycleHandle stream2 = marshaller.get(response);
+    assertNotNull(stream2);
+    stream2.close();
+    ResponseContentLifecycleHandle stream3 = marshaller.get(response);
+    assertNotNull(stream3);
+    stream3.close();
+  }
+
+  @Test
+  public void testParseOnSlowPath() throws IOException {
+    InputStream stream = createInputStream(response.toByteArray(), false);
+    ReadObjectResponseZeroCopyMessageMarshaller marshaller = createMarshaller();
+    ReadObjectResponse response = marshaller.parse(stream);
+    assertEquals(response, this.response);
+    ResponseContentLifecycleHandle stream2 = marshaller.get(response);
+    assertNotNull(stream2);
+    stream2.close();
+  }
+
+  @Test
+  public void testParseBrokenMessageOnFastPath() {
+    InputStream stream = createInputStream(dropLastOneByte(response.toByteArray()), true);
+    ReadObjectResponseZeroCopyMessageMarshaller marshaller = createMarshaller();
+    assertThrows(
+        StatusRuntimeException.class,
+        () -> {
+          marshaller.parse(stream);
+        });
+  }
+
+  @Test
+  public void testParseBrokenMessageOnSlowPath() {
+    InputStream stream = createInputStream(dropLastOneByte(response.toByteArray()), false);
+    ReadObjectResponseZeroCopyMessageMarshaller marshaller = createMarshaller();
+    assertThrows(
+        StatusRuntimeException.class,
+        () -> {
+          marshaller.parse(stream);
+        });
+  }
+
+  @Test
+  public void testResponseContentLifecycleHandle() throws IOException {
+    AtomicBoolean wasClosedCalled = new AtomicBoolean(false);
+    Closeable verifyClosed = () -> wasClosedCalled.set(true);
+
+    ResponseContentLifecycleHandle handle =
+        new ResponseContentLifecycleHandle(response, verifyClosed);
+    handle.close();
+
+    assertTrue(wasClosedCalled.get());
+
+    ResponseContentLifecycleHandle nullHandle = new ResponseContentLifecycleHandle(response, null);
+    nullHandle.close();
+    // No NullPointerException means test passes
+  }
+
+  @Test
+  public void testMarshallerClose_clean() throws IOException {
+    CloseAuditingInputStream stream1 =
+        CloseAuditingInputStream.of(createInputStream(response.toByteArray(), true));
+    CloseAuditingInputStream stream2 =
+        CloseAuditingInputStream.of(createInputStream(response.toByteArray(), true));
+    CloseAuditingInputStream stream3 =
+        CloseAuditingInputStream.of(createInputStream(response.toByteArray(), true));
+
+    ReadObjectResponseZeroCopyMessageMarshaller.closeAllStreams(
+        ImmutableList.of(stream1, stream2, stream3));
+
+    assertThat(stream1.closed).isTrue();
+    assertThat(stream2.closed).isTrue();
+    assertThat(stream3.closed).isTrue();
+  }
+
+  @SuppressWarnings("resource")
+  @Test
+  public void testMarshallerClose_multipleIoExceptions() {
+    CloseAuditingInputStream stream1 =
+        new CloseAuditingInputStream(null) {
+          @Override
+          void onClose() throws IOException {
+            throw new IOException("Kaboom stream1");
+          }
+        };
+    CloseAuditingInputStream stream2 =
+        new CloseAuditingInputStream(null) {
+          @Override
+          void onClose() throws IOException {
+            throw new IOException("Kaboom stream2");
+          }
+        };
+    CloseAuditingInputStream stream3 =
+        new CloseAuditingInputStream(null) {
+          @Override
+          void onClose() throws IOException {
+            throw new IOException("Kaboom stream3");
+          }
+        };
+
+    IOException ioException =
+        assertThrows(
+            IOException.class,
+            () ->
+                ReadObjectResponseZeroCopyMessageMarshaller.closeAllStreams(
+                    ImmutableList.of(stream1, stream2, stream3)));
+
+    assertThat(stream1.closed).isTrue();
+    assertThat(stream2.closed).isTrue();
+    assertThat(stream3.closed).isTrue();
+
+    assertThat(ioException).hasMessageThat().isEqualTo("Kaboom stream1");
+    List<String> messages =
+        Arrays.stream(ioException.getSuppressed())
+            .map(Throwable::getMessage)
+            .collect(Collectors.toList());
+    assertThat(messages).isEqualTo(ImmutableList.of("Kaboom stream2", "Kaboom stream3"));
+  }
+
+  private static class CloseAuditingInputStream extends FilterInputStream {
+
+    private boolean closed = false;
+
+    private CloseAuditingInputStream(InputStream in) {
+      super(in);
+    }
+
+    public static CloseAuditingInputStream of(InputStream in) {
+      return new CloseAuditingInputStream(in);
+    }
+
+    @Override
+    public void close() throws IOException {
+      closed = true;
+      onClose();
+      super.close();
+    }
+
+    void onClose() throws IOException {}
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionCommonSemanticsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionCommonSemanticsTest.java
new file mode 100644
index 00000000..dc02c3fb
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionCommonSemanticsTest.java
@@ -0,0 +1,223 @@
+/*
+ * 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.it;
+
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assume.assumeTrue;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.BlobWriteSessionConfig;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferAllocationStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartMetadataFieldDecorator;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.ITBlobWriteSessionCommonSemanticsTest.ParamsProvider;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.Parameterized;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = {Transport.HTTP, Transport.GRPC})
+@Parameterized(ParamsProvider.class)
+public final class ITBlobWriteSessionCommonSemanticsTest {
+
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Inject public Storage injectedStorage;
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  @Parameter public Params params;
+
+  private Storage storage;
+
+  @Before
+  public void setUp() throws Exception {
+    Path tmpDir = temporaryFolder.newFolder().toPath();
+    BlobWriteSessionConfig config = params.ctor.apply(tmpDir);
+
+    StorageOptions originalOptions = injectedStorage.getOptions();
+    StorageOptions newOptions = null;
+    try {
+      newOptions = originalOptions.toBuilder().setBlobWriteSessionConfig(config).build();
+    } catch (IllegalArgumentException e) {
+      assertThat(e).hasMessageThat().contains("not compatible with this");
+      assumeTrue(false);
+    }
+    storage = newOptions.getService();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @Test
+  public void closingAnOpenedSessionWithoutCallingWriteShouldMakeAnEmptyObject()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+
+    WritableByteChannel open = session.open();
+    open.close();
+    BlobInfo gen1 = session.getResult().get(1, TimeUnit.SECONDS);
+
+    // sometimes testbench will not define `.size = 0`, default it here if we get null
+    Long size = gen1.getSize();
+    if (size == null) {
+      size = 0L;
+    }
+    assertThat(size).isEqualTo(0);
+  }
+
+  @Test
+  public void attemptingToUseASessionWhichResultsInFailureShouldThrowAStorageException() {
+    // attempt to write to a bucket which we have not created
+    String badBucketName = bucket.getName() + "x";
+    BlobInfo info = BlobInfo.newBuilder(badBucketName, generator.randomObjectName()).build();
+
+    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+    StorageException se =
+        assertThrows(
+            StorageException.class,
+            () -> {
+              WritableByteChannel open = session.open();
+              open.close();
+            });
+
+    assertThat(se.getCode()).isEqualTo(404);
+  }
+
+  @Test
+  public void callingOpenIsOnlyAllowedOnce() throws Exception {
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+
+    WritableByteChannel open = session.open();
+    IllegalStateException se = assertThrows(IllegalStateException.class, session::open);
+
+    assertAll(() -> assertThat(se.getMessage()).contains("already open"));
+  }
+
+  @Test
+  public void getResultErrorsWhenTheSessionErrors() throws Exception {
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    byte[] helloWorld = "Hello World".getBytes(StandardCharsets.UTF_8);
+    storage.create(info, helloWorld, BlobTargetOption.doesNotExist());
+
+    BlobWriteSession session =
+        storage.blobWriteSession(
+            info,
+            // this precondition will result in failure
+            BlobWriteOption.doesNotExist());
+
+    try (WritableByteChannel open = session.open()) {
+      open.write(ByteBuffer.wrap(helloWorld));
+    } catch (StorageException se) {
+      assertThat(se.getCode()).isEqualTo(412);
+    } catch (IOException ioe) {
+      assertThat(ioe).hasCauseThat().isInstanceOf(StorageException.class);
+      StorageException se = (StorageException) ioe.getCause();
+      assertThat(se.getCode()).isEqualTo(412);
+    }
+
+    ExecutionException resultSe =
+        assertThrows(ExecutionException.class, () -> session.getResult().get(10, TimeUnit.SECONDS));
+
+    assertAll(
+        () -> assertThat(resultSe).hasCauseThat().isInstanceOf(StorageException.class),
+        () -> assertThat(((StorageException) resultSe.getCause()).getCode()).isEqualTo(412));
+  }
+
+  public static final class ParamsProvider implements ParametersProvider {
+    @Override
+    public ImmutableList<Params> parameters() {
+      final int _2MiB = 2 * 1024 * 1024;
+      final int _4MiB = 4 * 1024 * 1024;
+      return ImmutableList.of(
+          new Params("default", p -> BlobWriteSessionConfigs.getDefault()),
+          new Params("c!c.2MiB", p -> BlobWriteSessionConfigs.getDefault().withChunkSize(_2MiB)),
+          new Params("b!p.1", BlobWriteSessionConfigs::bufferToDiskThenUpload),
+          new Params("j!p.1", p -> BlobWriteSessionConfigs.journaling(ImmutableList.of(p))),
+          new Params(
+              "p!t.c&b.s*&p.4MiB&c.n&m.n",
+              p ->
+                  BlobWriteSessionConfigs.parallelCompositeUpload()
+                      .withExecutorSupplier(ExecutorSupplier.cachedPool())
+                      .withPartNamingStrategy(PartNamingStrategy.noPrefix())
+                      .withBufferAllocationStrategy(BufferAllocationStrategy.simple(_4MiB))
+                      .withPartCleanupStrategy(PartCleanupStrategy.never())
+                      .withPartMetadataFieldDecorator(PartMetadataFieldDecorator.noOp())),
+          new Params("d!c.2MiB", p -> BlobWriteSessionConfigs.bidiWrite().withBufferSize(_2MiB)));
+    }
+  }
+
+  public interface ParamsCtor {
+    BlobWriteSessionConfig apply(Path p) throws IOException;
+  }
+
+  public static final class Params {
+    private final String desc;
+    private final ParamsCtor ctor;
+
+    public Params(String desc, ParamsCtor ctor) {
+      this.desc = desc;
+      this.ctor = ctor;
+    }
+
+    @Override
+    public String toString() {
+      return desc;
+    }
+  }
+}
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 6cff6891..10ae174a 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
@@ -17,7 +17,6 @@
 package com.google.cloud.storage.it;
 
 import static com.google.common.truth.Truth.assertThat;
-import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
 
 import com.google.cloud.storage.BlobInfo;
@@ -25,11 +24,10 @@ 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.GrpcStorageOptions;
 import com.google.cloud.storage.HttpStorageOptions;
+import com.google.cloud.storage.JournalingBlobWriteSessionConfig;
 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.TransportCompatibility.Transport;
 import com.google.cloud.storage.it.runner.StorageITRunner;
@@ -37,12 +35,11 @@ import com.google.cloud.storage.it.runner.annotations.Backend;
 import com.google.cloud.storage.it.runner.annotations.CrossRun;
 import com.google.cloud.storage.it.runner.annotations.Inject;
 import com.google.cloud.storage.it.runner.registry.Generator;
-import java.io.IOException;
+import com.google.common.collect.ImmutableList;
 import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
-import java.util.concurrent.ExecutionException;
+import java.nio.file.Path;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -57,7 +54,6 @@ public final class ITBlobWriteSessionTest {
   @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
 
   @Inject public Storage storage;
-  @Inject public Transport transport;
 
   @Inject public BucketInfo bucket;
 
@@ -70,47 +66,43 @@ public final class ITBlobWriteSessionTest {
 
   @Test
   public void bufferToTempDirThenUpload() throws Exception {
-    StorageOptions options = null;
-    if (transport == Transport.GRPC) {
-      options =
-          ((GrpcStorageOptions) storage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
-              .build();
-    } else if (transport == Transport.HTTP) {
-      options =
-          ((HttpStorageOptions) storage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
-              .build();
-    }
-    assertWithMessage("unable to resolve options").that(options).isNotNull();
-    //noinspection DataFlowIssue
+    Path path = temporaryFolder.newFolder().toPath();
+    StorageOptions options =
+        storage
+            .getOptions()
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToDiskThenUpload(path))
+            .build();
     try (Storage s = options.getService()) {
       doTest(s);
     }
   }
 
+  @Test
+  @CrossRun.Exclude(transports = Transport.GRPC)
+  public void journalingNotSupportedByHttp() {
+    HttpStorageOptions.Builder builder = ((HttpStorageOptions) storage.getOptions()).toBuilder();
+
+    Path rootPath = temporaryFolder.getRoot().toPath();
+    JournalingBlobWriteSessionConfig journaling =
+        BlobWriteSessionConfigs.journaling(ImmutableList.of(rootPath));
+
+    IllegalArgumentException iae =
+        assertThrows(
+            IllegalArgumentException.class, () -> builder.setBlobWriteSessionConfig(journaling));
+
+    assertThat(iae).hasMessageThat().contains("HTTP transport");
+  }
+
   @Test
   public void overrideDefaultBufferSize() throws Exception {
-    StorageOptions options = null;
-    if (transport == Transport.GRPC) {
-      options =
-          ((GrpcStorageOptions) storage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(
-                  BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
-              .build();
-    } else if (transport == Transport.HTTP) {
-      options =
-          ((HttpStorageOptions) storage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(
-                  BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
-              .build();
-    }
-    assertWithMessage("unable to resolve options").that(options).isNotNull();
-    //noinspection DataFlowIssue
+    StorageOptions options =
+        (storage.getOptions())
+            .toBuilder()
+            .setBlobWriteSessionConfig(
+                BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
+            .build();
+
     try (Storage s = options.getService()) {
       doTest(s);
     }
@@ -119,47 +111,16 @@ public final class ITBlobWriteSessionTest {
   @Test
   @CrossRun.Exclude(transports = Transport.HTTP)
   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();
-
+    StorageOptions options =
+        (storage.getOptions())
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bidiWrite())
+            .build();
     try (Storage s = options.getService()) {
       doTest(s);
     }
   }
 
-  @Test
-  public void closingAnOpenedSessionWithoutCallingWriteShouldMakeAnEmptyObject()
-      throws IOException, ExecutionException, InterruptedException, TimeoutException {
-    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
-    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
-
-    WritableByteChannel open = session.open();
-    open.close();
-    BlobInfo gen1 = session.getResult().get(1, TimeUnit.SECONDS);
-
-    assertThat(gen1.getSize()).isEqualTo(0);
-  }
-
-  @Test
-  public void attemptingToOpenASessionWhichResultsInFailureShouldThrowAStorageException() {
-    // attempt to write to a bucket which we have not created
-    String badBucketName = bucket.getName() + "x";
-    BlobInfo info = BlobInfo.newBuilder(badBucketName, generator.randomObjectName()).build();
-
-    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
-    StorageException se = assertThrows(StorageException.class, () -> session.open().close());
-
-    assertThat(se.getCode()).isEqualTo(404);
-    assertThat(se).hasMessageThat().contains(badBucketName);
-  }
-
   private void doTest(Storage underTest) throws Exception {
     BlobWriteSession sess =
         underTest.blobWriteSession(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java
index 4526a09f..70e369f3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java
@@ -24,7 +24,6 @@ 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.GrpcStorageOptions;
 import com.google.cloud.storage.JournalingBlobWriteSessionConfig;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
@@ -75,7 +74,9 @@ public final class ITJournalingBlobWriteSessionConfigTest {
     JournalingBlobWriteSessionConfig journaling =
         BlobWriteSessionConfigs.journaling(ImmutableList.of(tempDir));
     journalingStorage =
-        ((GrpcStorageOptions.Builder) this.storage.getOptions().toBuilder())
+        this.storage
+            .getOptions()
+            .toBuilder()
             .setBlobWriteSessionConfig(journaling)
             .build()
             .getService();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
index 3352a92a..9b7d630f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
@@ -18,7 +18,6 @@ package com.google.cloud.storage.it;
 
 import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.common.truth.Truth.assertThat;
-import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
 
 import com.google.api.gax.paging.Page;
@@ -30,8 +29,6 @@ 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.GrpcStorageOptions;
-import com.google.cloud.storage.HttpStorageOptions;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferAllocationStrategy;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
@@ -119,22 +116,8 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
             // let our fixtures take care of cleaning things
             .withPartCleanupStrategy(PartCleanupStrategy.never());
 
-    StorageOptions storageOptions = null;
-    if (transport == Transport.GRPC) {
-      storageOptions =
-          ((GrpcStorageOptions) injectedStorage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(pcu)
-              .build();
-    } else if (transport == Transport.HTTP) {
-      storageOptions =
-          ((HttpStorageOptions) injectedStorage.getOptions())
-              .toBuilder()
-              .setBlobWriteSessionConfig(pcu)
-              .build();
-    }
-    assertWithMessage("unable to resolve options").that(storageOptions).isNotNull();
-    //noinspection DataFlowIssue
+    StorageOptions storageOptions =
+        injectedStorage.getOptions().toBuilder().setBlobWriteSessionConfig(pcu).build();
     storage = storageOptions.getService();
     rand = new Random();
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
index 4f345af0..5c1feed6 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
@@ -74,6 +74,8 @@ public class ITTransferManagerTest {
   private static final Comparator<DownloadResult> comp2 =
       Comparator.comparing(DownloadResult::getInput, comp);
 
+  private static final long CHUNK_THRESHOLD = 2L * 1024 * 1024;
+
   @Inject public Storage storage;
   @Inject public BucketInfo bucket;
   @Inject public Generator generator;
@@ -106,8 +108,8 @@ public class ITTransferManagerTest {
         writeChannel.write(content);
       }
     }
-    long size = 2L * 1024 * 1024;
-    size = size + 100L;
+    // We make this size just a bit bigger than the threshold.
+    long size = CHUNK_THRESHOLD + 100L;
     ByteBuffer chunkedContent = DataGenerator.base64Characters().genByteBuffer(size);
     try (WriteChannel writeChannel = storage.writer(blobInfoChunking)) {
       writeChannel.write(chunkedContent);
@@ -341,6 +343,56 @@ public class ITTransferManagerTest {
     }
   }
 
+  @Test
+  public void uploadFilesAllowPCU() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowParallelCompositeUpload(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    long size = CHUNK_THRESHOLD + 100L;
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, size)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucket.getName()).build();
+      UploadJob job =
+          transferManager.uploadFiles(
+              Collections.singletonList(tmpFile.getPath()), parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+    }
+  }
+
+  @Test
+  public void uploadFilesAllowMultiplePCUAndSmallerFiles() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions())
+            .toBuilder()
+            .setAllowParallelCompositeUpload(true)
+            .setPerWorkerBufferSize(128 * 1024)
+            .build();
+    long largeFileSize = CHUNK_THRESHOLD + 100L;
+    long smallFileSize = CHUNK_THRESHOLD - 100L;
+    try (TransferManager transferManager = config.getService();
+        TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, largeFileSize);
+        TmpFile tmpfile2 = DataGenerator.base64Characters().tempFile(baseDir, largeFileSize);
+        TmpFile tmpFile3 = DataGenerator.base64Characters().tempFile(baseDir, smallFileSize)) {
+      ParallelUploadConfig parallelUploadConfig =
+          ParallelUploadConfig.newBuilder().setBucketName(bucket.getName()).build();
+      List<Path> files =
+          ImmutableList.of(tmpFile.getPath(), tmpfile2.getPath(), tmpFile3.getPath());
+      UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
+      List<UploadResult> uploadResults = job.getUploadResults();
+      assertThat(uploadResults).hasSize(3);
+      assertThat(
+              uploadResults.stream()
+                  .filter(result -> result.getStatus() == TransferStatus.SUCCESS)
+                  .collect(Collectors.toList()))
+          .hasSize(3);
+    }
+  }
+
   @Test
   public void downloadNonexistentBucket() throws Exception {
     TransferManagerConfig config =
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
index e9ef86c3..da89ada5 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerConfigTestingInstances.java
@@ -28,9 +28,8 @@ public final class TransferManagerConfigTestingInstances {
   public static TransferManagerConfig defaults(StorageOptions options) {
     return TransferManagerConfig.newBuilder()
         .setAllowDivideAndConquerDownload(false)
-        .setMaxWorkers(1)
+        .setMaxWorkers(5)
         .setPerWorkerBufferSize(512 * 1024)
-        .setQos(DefaultQos.of(2 * 1024 * 1024))
         .setStorageOptions(options)
         .build();
   }
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
new file mode 100644
index 00000000..e43b3a16
--- /dev/null
+++ b/samples/snippets/src/test/java/com/example/storage/transfermanager/ITTransferManagerSamples.java
@@ -0,0 +1,119 @@
+/*
+ * 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;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+
+public class ITTransferManagerSamples {
+  private static final String BUCKET = RemoteStorageHelper.generateBucketName();
+  private static Storage storage;
+  private static List<BlobInfo> blobs;
+  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();
+  @Rule public final TemporaryFolder tmpDirectory = new TemporaryFolder();
+
+  @BeforeClass
+  public static void beforeClass() {
+    RemoteStorageHelper helper = RemoteStorageHelper.create();
+    storage = helper.getOptions().getService();
+    storage.create(BucketInfo.of(BUCKET));
+    blobs = Arrays.asList(BlobInfo.newBuilder(BUCKET, "blob1").build(),
+        BlobInfo.newBuilder(BUCKET, "blob2").build(),
+        BlobInfo.newBuilder(BUCKET, "blob3").build());
+    for (BlobInfo blob : blobs) {
+      storage.create(blob);
+    }
+
+  }
+  
+  @Test
+  public void uploadFiles() throws Exception {
+    File tmpFile = File.createTempFile("file", ".txt");
+    File tmpFile2 = File.createTempFile("file2", ".txt");
+    File tmpFile3 = File.createTempFile("file3", ".txt");
+    List<Path> files =
+        ImmutableList.of(tmpFile.toPath(), tmpFile2.toPath(), tmpFile3.toPath());
+    UploadMany.uploadManyFiles(BUCKET, files);
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("file")).isTrue();
+    assertThat(snippetOutput.contains("file2")).isTrue();
+    assertThat(snippetOutput.contains("file3")).isTrue();
+  }
+
+  @Test
+  public void uploadDirectory() throws IOException {
+    File tmpFile = tmpDirectory.newFile("fileDirUpload.txt");
+    File tmpFile2 = tmpDirectory.newFile("fileDirUpload2.txt");
+    File tmpFile3 = tmpDirectory.newFile("fileDirUpload3.txt");
+    UploadDirectory.uploadDirectoryContents(BUCKET, tmpDirectory.getRoot().toPath());
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("fileDirUpload.txt")).isTrue();
+    assertThat(snippetOutput.contains("fileDirUpload2.txt")).isTrue();
+    assertThat(snippetOutput.contains("fileDirUpload3.txt")).isTrue();
+  }
+
+  @Test
+  public void downloadBucket() {
+    String downloadFullBucketName = RemoteStorageHelper.generateBucketName();
+    storage.create(BucketInfo.of(downloadFullBucketName));
+    List<BlobInfo> bucketBlobs = Arrays.asList(
+        BlobInfo.newBuilder(downloadFullBucketName, "bucketb1").build(),
+        BlobInfo.newBuilder(downloadFullBucketName, "bucketb2").build(),
+        BlobInfo.newBuilder(downloadFullBucketName, "bucketb3").build());
+    for (BlobInfo blob : bucketBlobs) {
+      storage.create(blob);
+    }
+    DownloadBucket
+        .downloadBucketContents(PROJECT_ID, downloadFullBucketName, tmp.getRoot().toPath());
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("bucketb1")).isTrue();
+    assertThat(snippetOutput.contains("bucketb2")).isTrue();
+    assertThat(snippetOutput.contains("bucketb3")).isTrue();
+  }
+
+  @Test
+  public void downloadFiles() {
+    DownloadMany.downloadManyBlobs(BUCKET, blobs, tmp.getRoot().toPath());
+    String snippetOutput = stdOutCaptureRule.getCapturedOutputAsUtf8String();
+    assertThat(snippetOutput.contains("blob1")).isTrue();
+    assertThat(snippetOutput.contains("blob2")).isTrue();
+    assertThat(snippetOutput.contains("blob3")).isTrue();
+  }
+
+}