Repository: googleapis/java-storage
Pull request #2492 (2024-04-08 17:51:55)

docs: Create Samples for transfer manager
Samples for the upload many operations only. Will do chunked once Parallel Composite Upload is complete.


snippet-bot[bot]: <!-- probot comment [11299897]-->
Here is the summary of changes.
<details>
  <summary>You are about to add 4 region tags.</summary>

  - [samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadBucket.java:19](https://github.com/googleapis/java-storage/blob/393515fe28487eecf5d5f70ccc4d61de5bb10a21/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadBucket.java#L19), tag `storage_transfer_manager_download_bucket`
- [samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadMany.java:19](https://github.com/googleapis/java-storage/blob/393515fe28487eecf5d5f70ccc4d61de5bb10a21/samples/snippets/src/main/java/com/example/storage/transfermanager/DownloadMany.java#L19), tag `storage_transfer_manager_download_many`
- [samples/snippets/src/main/java/com/example/storage/transfermanager/UploadDirectory.java:19](https://github.com/googleapis/java-storage/blob/393515fe28487eecf5d5f70ccc4d61de5bb10a21/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadDirectory.java#L19), tag `storage_transfer_manager_upload_directory`
- [samples/snippets/src/main/java/com/example/storage/transfermanager/UploadMany.java:19](https://github.com/googleapis/java-storage/blob/393515fe28487eecf5d5f70ccc4d61de5bb10a21/samples/snippets/src/main/java/com/example/storage/transfermanager/UploadMany.java#L19), tag `storage_transfer_manager_upload_many`

</details>

---
This comment is generated by [snippet-bot](https://github.com/apps/snippet-bot).
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add `snippet-bot:force-run` label or use the checkbox below:
- [ ] Refresh this comment


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/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/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/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/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/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/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 4abe9494..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,12 +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;
@@ -39,13 +36,10 @@ 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 com.google.common.collect.ImmutableList;
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.Path;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -60,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;
 
@@ -73,22 +66,13 @@ 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);
     }
@@ -112,24 +96,13 @@ public final class ITBlobWriteSessionTest {
 
   @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);
     }
@@ -138,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();
+  }
+
+}