Repository: googleapis/java-storage
Pull request #2239 (2023-09-29 19:06:00)

feat: add @BetaApi BlobWriteSession#parallelCompositeUpload
Expose public api to allow selection and configuration of new Parallel Composite Upload BlobWriteSession strategy


Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
index 76296e9b..e07043e5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
@@ -18,6 +18,8 @@ package com.google.cloud.storage;
 
 import com.google.api.core.BetaApi;
 import com.google.cloud.storage.GrpcStorageOptions.GrpcStorageDefaults;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.common.collect.ImmutableList;
 import java.io.IOException;
@@ -108,6 +110,10 @@ import java.util.Collection;
  *       retryable error query the offset of the Resumable Upload Session, then open the recovery
  *       file from the offset and transmit the bytes to Cloud Storage.
  *     </td>
+ *     <td>
+ *       Opening the stream for upload will be retried up to the limitations specified in {@link StorageOptions#getRetrySettings()}
+ *       All bytes are buffered to disk and allow for recovery from any arbitrary offset.
+ *     </td>
  *     <td>gRPC</td>
  *     <td><a href="https://cloud.google.com/storage/docs/resumable-uploads">Resumable Upload</a></td>
  *     <td>
@@ -128,6 +134,90 @@ import java.util.Collection;
  *       </ol>
  *     </td>
  *   </tr>
+ *   <tr>
+ *     <td>Parallel Composite Upload</td>
+ *     <td>{@link #parallelCompositeUpload()}</td>
+ *     <td>
+ *       Break the stream of bytes into smaller part objects uploading each part in parallel. Then
+ *       composing the parts together to make the ultimate object.
+ *     </td>
+ *     <td>
+ *       Automatic retires will be applied for the following:
+ *       <ol>
+ *         <li>Creation of each individual part</li>
+ *         <li>Performing an intermediary compose</li>
+ *         <li>Performing a delete to cleanup each part and intermediary compose object</li>
+ *       </ol>
+ *
+ *       Retrying the creation of the final object is contingent upon if an appropriate precondition
+ *       is supplied when calling {@link Storage#blobWriteSession(BlobInfo, BlobWriteOption...)}.
+ *       Either {@link BlobTargetOption#doesNotExist()} or {@link Storage.BlobTargetOption#generationMatch(long)}
+ *       should be specified in order to make the final request idempotent.
+ *       <p>Each operation will be retried up to the limitations specified in {@link StorageOptions#getRetrySettings()}
+ *     </td>
+ *     <td>gRPC</td>
+ *     <td>
+ *       <ul>
+ *         <li><a href="https://cloud.google.com/storage/docs/parallel-composite-uploads">Parallel composite uploads</a></li>
+ *         <li><a href="https://cloud.google.com/storage/docs/uploading-objects-from-memory">Direct uploads</a></li>
+ *         <li><a href="https://cloud.google.com/storage/docs/composite-objects">Compose</a></li>
+ *         <li><a href="https://cloud.google.com/storage/docs/deleting-objects">Object delete</a></li>
+ *       </ul>
+ *     </td>
+ *     <td>
+ *       <ol>
+ *         <li>
+ *           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>An illustrative example. Upload a 5GiB object using 64MiB as the max size per part.
+ *           <ol>
+ *             <li>80 Parts will be created (Class A)</li>
+ *             <li>3 compose calls will be performed (Class A)</li>
+ *             <li>Delete 80 Parts along with 2 intermediary Compose objects (Free tier as long as {@code STANDARD} class)</li>
+ *           </ol>
+ *
+ *           Once the parts and intermediary compose objects are deleted, there will be no storage charges related to those temporary objects.
+ *         </li>
+ *         <li>
+ *           The service account/credentials used to perform the parallel composite upload require
+ *           <a href="https://cloud.google.com/storage/docs/access-control/iam-permissions#object_permissions">{@code storage.objects.delete}</a>
+ *           in order to cleanup the temporary part and intermediary compose objects.
+ *           <p><i>To handle handle part and intermediary compose object deletion out of band</i>
+ *           passing {@link PartCleanupStrategy#never()} to {@link ParallelCompositeUploadBlobWriteSessionConfig#withPartCleanupStrategy(PartCleanupStrategy)}
+ *           will prevent automatic cleanup.
+ *         </li>
+ *         <li>
+ *           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.
+ *         </li>
+ *         <li>
+ *           A failed upload can leave part and intermediary compose objects behind which will count
+ *           as storage usage, and you will be billed for it.
+ *           <p>By default if an upload fails, an attempt to cleanup the part and intermediary compose
+ *           will be made. However if the program were to crash there is no means for the client to
+ *           perform the cleanup.
+ *           <p>Every part and intermediary compose object will be created with a name which ends in
+ *           {@code .part}. An Object Lifecycle Management rule can be setup on your bucket to automatically
+ *           cleanup objects with the suffix after some period of time. See
+ *           <a href="https://cloud.google.com/storage/docs/lifecycle">Object Lifecycle Management</a>
+ *           for full details and a guide on how to setup a <a href="https://cloud.google.com/storage/docs/lifecycle#delete">Delete</a>
+ *           rule with a <a href="https://cloud.google.com/storage/docs/lifecycle#matchesprefix-suffix">suffix match</a> condition.
+ *         </li>
+ *         <li>
+ *           Using parallel composite uploads are not a one size fits all solution. They have very
+ *           real overhead until uploading a large enough object. The inflection point is dependent
+ *           upon many factors, and there is no one size fits all value. You will need to experiment
+ *           with your deployment and workload to determine if parallel composite uploads are useful
+ *           to you.
+ *         </li>
+ *       </ol>
+ *     </td>
+ *   </tr>
  * </table>
  *
  * @see BlobWriteSessionConfig
@@ -219,4 +309,19 @@ public final class BlobWriteSessionConfigs {
   public static JournalingBlobWriteSessionConfig journaling(Collection<Path> paths) {
     return new JournalingBlobWriteSessionConfig(ImmutableList.copyOf(paths), false);
   }
+
+  /**
+   * Create a new {@link BlobWriteSessionConfig} which will perform a <a
+   * href="https://cloud.google.com/storage/docs/parallel-composite-uploads">Parallel Composite
+   * Upload</a> by breaking the stream into parts and composing the parts together to make the
+   * ultimate object.
+   *
+   * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+   * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  public static ParallelCompositeUploadBlobWriteSessionConfig parallelCompositeUpload() {
+    return ParallelCompositeUploadBlobWriteSessionConfig.withDefaults();
+  }
 }
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 e8d6130f..16963e4d 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
@@ -1815,7 +1815,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
     Set<StatusCode.Code> codes =
         resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
+    GrpcCallContext grpcCallContext =
+        opts.grpcMetadataMapper().apply(Retrying.newCallContext().withRetryableCodes(codes));
     return ResumableMedia.gapic()
         .read()
         .byteChannel(storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext))
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
index d2ed5b43..aa86de36 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfig.java
@@ -19,14 +19,23 @@ package com.google.cloud.storage;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
+import com.google.api.core.SettableApiFuture;
+import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
+import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.MetadataField.PartRange;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.hash.HashCode;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.security.SecureRandom;
@@ -41,16 +50,70 @@ import javax.annotation.concurrent.Immutable;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 /**
- * Immutable config builder for a Parallel Composite Upload
+ * Immutable config builder to configure BlobWriteSession instances to perform Parallel Composite
+ * Uploads.
  *
- * @see <a
- *     href="https://cloud.google.com/storage/docs/composing-objects">https://cloud.google.com/storage/docs/composing-objects</a>
+ * <p>Parallel Composite Uploads can yield higher throughput when uploading large objects. However,
+ * there are some things which must be kept in mind when choosing to use this strategy.
+ *
+ * <ol>
+ *   <li>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>An illustrative example. Upload a 5GiB object using 64MiB as the max size per part. <br>
+ *       <ol>
+ *         <li>80 Parts will be created (Class A)
+ *         <li>3 compose calls will be performed (Class A)
+ *         <li>Delete 80 Parts along with 2 intermediary Compose objects (Free tier as long as
+ *             {@code STANDARD} class)
+ *       </ol>
+ *       Once the parts and intermediary compose objects are deleted, there will be no storage
+ *       charges related to those temporary objects.
+ *   <li>The service account/credentials used to perform the parallel composite upload require <a
+ *       href="https://cloud.google.com/storage/docs/access-control/iam-permissions#object_permissions">{@code
+ *       storage.objects.delete}</a> in order to cleanup the temporary part and intermediary compose
+ *       objects. <br>
+ *       <i>To handle handle part and intermediary compose object deletion out of band</i> passing
+ *       {@link PartCleanupStrategy#never()} to {@link
+ *       ParallelCompositeUploadBlobWriteSessionConfig#withPartCleanupStrategy(PartCleanupStrategy)}
+ *       will prevent automatic cleanup.
+ *   <li>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.
+ *   <li>A failed upload can leave part and intermediary compose objects behind which will count as
+ *       storage usage, and you will be billed for it. <br>
+ *       By default if an upload fails, an attempt to cleanup the part and intermediary compose will
+ *       be made. However if the program were to crash there is no means for the client to perform
+ *       the cleanup. <br>
+ *       Every part and intermediary compose object will be created with a name which ends in {@code
+ *       .part}. An Object Lifecycle Management rule can be setup on your bucket to automatically
+ *       cleanup objects with the suffix after some period of time. See <a
+ *       href="https://cloud.google.com/storage/docs/lifecycle">Object Lifecycle Management</a> for
+ *       full details and a guide on how to setup a <a
+ *       href="https://cloud.google.com/storage/docs/lifecycle#delete">Delete</a> rule with a <a
+ *       href="https://cloud.google.com/storage/docs/lifecycle#matchesprefix-suffix">suffix
+ *       match</a> condition.
+ *   <li>Using parallel composite uploads are not a a one size fits all solution. They have very
+ *       real overhead until uploading a large enough object. The inflection point is dependent upon
+ *       many factors, and there is no one size fits all value. You will need to experiment with
+ *       your deployment and workload to determine if parallel composite uploads are useful to you.
+ * </ol>
+ *
+ * <p>In general if you object sizes are smaller than several hundred megabytes it is unlikely
+ * parallel composite uploads will be beneficial to overall throughput.
+ *
+ * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+ * @see BlobWriteSessionConfigs#parallelCompositeUpload()
+ * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
  * @see <a
  *     href="https://cloud.google.com/storage/docs/parallel-composite-uploads">https://cloud.google.com/storage/docs/parallel-composite-uploads</a>
+ * @since 2.28.0 This new api is in preview and is subject to breaking changes.
  */
 @Immutable
 @BetaApi
-final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig {
+public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessionConfig {
 
   private static final int MAX_PARTS_PER_COMPOSE = 32;
   private final int maxPartsPerCompose;
@@ -86,6 +149,14 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
         partCleanupStrategy);
   }
 
+  /**
+   * Specify a specific executor supplier where work will be submitted when performing a parallel
+   * composite upload.
+   *
+   * <p><i>Default: </i> {@link ExecutorSupplier#cachedPool()}
+   *
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   public ParallelCompositeUploadBlobWriteSessionConfig withExecutorSupplier(
       ExecutorSupplier executorSupplier) {
@@ -98,6 +169,14 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
         partCleanupStrategy);
   }
 
+  /**
+   * Specify a specific buffering strategy which will dictate how buffers are allocated and used
+   * when performing a parallel composite upload.
+   *
+   * <p><i>Default: </i> {@link BufferStrategy#simple(int) BufferStrategy#simple(16MiB)}
+   *
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   public ParallelCompositeUploadBlobWriteSessionConfig withBufferStrategy(
       BufferStrategy bufferStrategy) {
@@ -110,6 +189,14 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
         partCleanupStrategy);
   }
 
+  /**
+   * Specify a specific naming strategy which will dictate how individual part and intermediary
+   * compose objects will be named when performing a parallel composite upload.
+   *
+   * <p><i>Default: </i> {@link PartNamingStrategy#noPrefix()}
+   *
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   public ParallelCompositeUploadBlobWriteSessionConfig withPartNamingStrategy(
       PartNamingStrategy partNamingStrategy) {
@@ -122,6 +209,14 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
         partCleanupStrategy);
   }
 
+  /**
+   * Specify a specific cleanup strategy which will dictate what cleanup operations are performed
+   * automatically when performing a parallel composite upload.
+   *
+   * <p><i>Default: </i> {@link PartCleanupStrategy#always()}
+   *
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   public ParallelCompositeUploadBlobWriteSessionConfig withPartCleanupStrategy(
       PartCleanupStrategy partCleanupStrategy) {
@@ -135,7 +230,7 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
   }
 
   @BetaApi
-  static ParallelCompositeUploadBlobWriteSessionConfig of() {
+  static ParallelCompositeUploadBlobWriteSessionConfig withDefaults() {
     return new ParallelCompositeUploadBlobWriteSessionConfig(
         MAX_PARTS_PER_COMPOSE,
         ExecutorSupplier.cachedPool(),
@@ -149,20 +244,45 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
   WriterFactory createFactory(Clock clock) throws IOException {
     Executor executor = executorSupplier.get();
     BufferHandlePool bufferHandlePool = bufferStrategy.get();
-    throw new IllegalStateException("Not yet implemented");
+    return new ParallelCompositeUploadWriterFactory(clock, executor, bufferHandlePool);
   }
 
+  /**
+   * A strategy which dictates how buffers are to be used for individual parts. The chosen strategy
+   * will apply to all instances of {@link BlobWriteSession} created from a single instance of
+   * {@link Storage}.
+   *
+   * @see #withBufferStrategy(BufferStrategy)
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   @Immutable
   public abstract static class BufferStrategy extends Factory<BufferHandlePool> {
 
     private BufferStrategy() {}
 
+    /**
+     * Create a buffer strategy which will rely upon standard garbage collection. Each buffer will
+     * be used once and then garbage collected.
+     *
+     * @param capacity the number of bytes each buffer should be
+     * @see #withBufferStrategy(BufferStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static BufferStrategy simple(int capacity) {
       return new SimpleBufferStrategy(capacity);
     }
 
+    /**
+     * Create a buffer strategy which will have a fixed size pool of buffers. Each buffer will be
+     * lazily allocated.
+     *
+     * @param bufferCount the number of buffers the pool will be
+     * @param bufferCapacity the number of bytes each buffer should be
+     * @see #withBufferStrategy(BufferStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static BufferStrategy fixedPool(int bufferCount, int bufferCapacity) {
       return new FixedBufferStrategy(bufferCount, bufferCapacity);
@@ -199,6 +319,13 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
     }
   }
 
+  /**
+   * Class which will be used to supply an Executor where work will be submitted when performing a
+   * parallel composite upload.
+   *
+   * @see #withExecutorSupplier(ExecutorSupplier)
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   @Immutable
   public abstract static class ExecutorSupplier extends Factory<Executor> {
@@ -206,6 +333,12 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
 
     private ExecutorSupplier() {}
 
+    /**
+     * Create a cached thread pool for submitting work
+     *
+     * @see #withExecutorSupplier(ExecutorSupplier)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static ExecutorSupplier cachedPool() {
       return new ExecutorSupplier() {
@@ -217,6 +350,13 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
       };
     }
 
+    /**
+     * Create a fixed size thread pool for submitting work
+     *
+     * @param poolSize the number of threads in the pool
+     * @see #withExecutorSupplier(ExecutorSupplier)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static ExecutorSupplier fixedPool(int poolSize) {
       return new ExecutorSupplier() {
@@ -228,6 +368,16 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
       };
     }
 
+    /**
+     * Wrap an existing executor instance which will be used for submitting work
+     *
+     * <p><b><i>Choosing to use this supplier type will make your instance of {@link StorageOptions}
+     * unable to be serialized.</i></b>
+     *
+     * @param executor the executor to use
+     * @see #withExecutorSupplier(ExecutorSupplier)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static ExecutorSupplier useExecutor(Executor executor) {
       return new SuppliedExecutorSupplier(executor);
@@ -256,6 +406,13 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
     }
   }
 
+  /**
+   * A naming strategy which will be used to generate a name for a part or intermediary compose
+   * object.
+   *
+   * @see #withPartNamingStrategy(PartNamingStrategy)
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   @Immutable
   public abstract static class PartNamingStrategy {
@@ -285,12 +442,52 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
 
     protected abstract String fmtFields(String randomKey, String nameDigest, String partRange);
 
+    /**
+     * Default strategy in which no stable prefix is defined.
+     *
+     * <p>General format is
+     *
+     * <pre><code>
+     *   {randomKeyDigest};{objectInfoDigest};{partIndex}.part
+     * </code></pre>
+     *
+     * <p>{@code {objectInfoDigest}} will be fixed for an individual {@link BlobWriteSession}.
+     *
+     * <p><b><i>NOTE:</i></b>The way in which both {@code randomKeyDigest} and {@code
+     * objectInfoDigest} are generated is undefined and subject to change at any time.
+     *
+     * @see #withPartNamingStrategy(PartNamingStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static PartNamingStrategy noPrefix() {
       SecureRandom rand = new SecureRandom();
       return new NoPrefix(rand);
     }
 
+    /**
+     * Strategy in which an explicit stable prefix is present on each part and intermediary compose
+     * object.
+     *
+     * <p>General format is
+     *
+     * <pre><code>
+     *   {prefixPattern}/{randomKeyDigest};{objectInfoDigest};{partIndex}.part
+     * </code></pre>
+     *
+     * <p>{@code {objectInfoDigest}} will be fixed for an individual {@link BlobWriteSession}.
+     *
+     * <p><b><i>NOTE:</i></b>The way in which both {@code randomKeyDigest} and {@code
+     * objectInfoDigest} are generated is undefined and subject to change at any time.
+     *
+     * <p>Care must be taken when choosing to specify a stable prefix as this can create hotspots in
+     * the keyspace for object names. See <a
+     * href="https://cloud.google.com/storage/docs/request-rate#naming-convention">Object Naming
+     * Convention Guidelines</a> for more details.
+     *
+     * @see #withPartNamingStrategy(PartNamingStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static PartNamingStrategy prefix(String prefixPattern) {
       checkNotNull(prefixPattern, "prefixPattern must be non null");
@@ -342,6 +539,13 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
     }
   }
 
+  /**
+   * A cleanup strategy which will dictate what cleanup operations are performed automatically when
+   * performing a parallel composite upload.
+   *
+   * @see #withPartCleanupStrategy(PartCleanupStrategy)
+   * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+   */
   @BetaApi
   @Immutable
   public static class PartCleanupStrategy {
@@ -364,17 +568,47 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
     /**
      * If an unrecoverable error is encountered, define whether to attempt to delete any object
      * parts already uploaded.
+     *
+     * <p><i>Default:</i> {@code true}
+     *
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
      */
     @BetaApi
-    public PartCleanupStrategy withDeleteOnError(boolean deleteOnError) {
+    PartCleanupStrategy withDeleteOnError(boolean deleteOnError) {
       return new PartCleanupStrategy(deleteParts, deleteOnError);
     }
 
+    /**
+     * Cleanup strategy which will always attempt to clean up part and intermediary compose objects
+     * either on success or on error.
+     *
+     * @see #withPartCleanupStrategy(PartCleanupStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static PartCleanupStrategy always() {
       return new PartCleanupStrategy(true, true);
     }
 
+    /**
+     * Cleanup strategy which will only attempt to clean up parts and intermediary compose objects
+     * either on success.
+     *
+     * @see #withPartCleanupStrategy(PartCleanupStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public static PartCleanupStrategy onlyOnSuccess() {
+      return new PartCleanupStrategy(true, false);
+    }
+
+    /**
+     * Cleanup strategy which will never attempt to clean up parts or intermediary compose objects
+     * either on success or on error.
+     *
+     * @see #withPartCleanupStrategy(PartCleanupStrategy)
+     * @since 2.28.0 This new api is in preview and is subject to breaking changes.
+     */
     @BetaApi
     public static PartCleanupStrategy never() {
       return new PartCleanupStrategy(false, false);
@@ -384,4 +618,65 @@ final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWriteSessi
   private abstract static class Factory<T> {
     abstract T get();
   }
+
+  private class ParallelCompositeUploadWriterFactory implements WriterFactory {
+
+    private final Clock clock;
+    private final Executor executor;
+    private final BufferHandlePool bufferHandlePool;
+
+    private ParallelCompositeUploadWriterFactory(
+        Clock clock, Executor executor, BufferHandlePool bufferHandlePool) {
+      this.clock = clock;
+      this.executor = executor;
+      this.bufferHandlePool = bufferHandlePool;
+    }
+
+    @Override
+    public WritableByteChannelSession<?, BlobInfo> writeSession(
+        StorageInternal s,
+        BlobInfo info,
+        Opts<ObjectTargetOpt> opts,
+        Decoder<WriteObjectResponse, BlobInfo> d) {
+      return new PCUSession(s, info, opts);
+    }
+
+    private final class PCUSession
+        implements WritableByteChannelSession<BufferedWritableByteChannel, BlobInfo> {
+
+      private final SettableApiFuture<BlobInfo> result;
+      private final StorageInternal storageInternal;
+      private final BlobInfo info;
+      private final Opts<ObjectTargetOpt> opts;
+
+      private PCUSession(
+          StorageInternal storageInternal, BlobInfo info, Opts<ObjectTargetOpt> opts) {
+        this.storageInternal = storageInternal;
+        this.info = info;
+        this.opts = opts;
+        result = SettableApiFuture.create();
+      }
+
+      @Override
+      public ApiFuture<BufferedWritableByteChannel> openAsync() {
+        ParallelCompositeUploadWritableByteChannel channel =
+            new ParallelCompositeUploadWritableByteChannel(
+                bufferHandlePool,
+                executor,
+                partNamingStrategy,
+                partCleanupStrategy,
+                maxPartsPerCompose,
+                result,
+                storageInternal,
+                info,
+                opts);
+        return ApiFutures.immediateFuture(channel);
+      }
+
+      @Override
+      public ApiFuture<BlobInfo> getResult() {
+        return result;
+      }
+    }
+  }
 }


Test patch:
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
new file mode 100644
index 00000000..72a07dd5
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
@@ -0,0 +1,248 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.it;
+
+import static com.google.cloud.storage.TestUtils.xxd;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.cloud.kms.v1.CryptoKey;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.GrpcStorageOptions;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+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.TransportCompatibility.Transport;
+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.BucketFixture;
+import com.google.cloud.storage.it.runner.annotations.BucketType;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.cloud.storage.it.runner.annotations.StorageFixture;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.it.runner.registry.KmsFixture;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.WritableByteChannel;
+import java.security.Key;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@SingleBackend(Backend.PROD)
+public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
+
+  private static final int _1MiB = 1024 * 1024;
+  private static ExecutorService exec;
+
+  @Inject public BucketInfo bucket;
+
+  @Inject
+  @BucketFixture(BucketType.REQUESTER_PAYS)
+  public BucketInfo rpBucket;
+
+  @Inject
+  @StorageFixture(Transport.GRPC)
+  public Storage injectedStorage;
+
+  @Inject public Generator generator;
+  @Inject public KmsFixture kmsFixture;
+
+  // configured Storage with the PCU config
+  private Storage storage;
+  private Random rand;
+
+  @BeforeClass
+  public static void beforeClass() {
+    ThreadFactory threadFactory =
+        new ThreadFactoryBuilder().setDaemon(true).setNameFormat("test-pcuwbct-%d").build();
+    exec = Executors.newCachedThreadPool(threadFactory);
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    GrpcStorageOptions storageOptions =
+        ((GrpcStorageOptions) injectedStorage.getOptions())
+            .toBuilder()
+            .setBlobWriteSessionConfig(
+                BlobWriteSessionConfigs.parallelCompositeUpload()
+                    .withExecutorSupplier(ExecutorSupplier.useExecutor(exec))
+                    // deinfe a max part size that is fairly small to aid in test speed
+                    .withBufferStrategy(BufferStrategy.simple(_1MiB))
+                    .withPartNamingStrategy(PartNamingStrategy.prefix("prefix-a"))
+                    // let our fixtures take care of cleaning things up
+                    .withPartCleanupStrategy(PartCleanupStrategy.never()))
+            .build();
+    storage = storageOptions.getService();
+    rand = new Random();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    if (exec != null) {
+      exec.shutdownNow();
+    }
+  }
+
+  @Test
+  public void errorRaisedByMethodAndFutureResult() throws IOException {
+
+    BlobInfo info =
+        BlobInfo.newBuilder(bucket.getName() + "x", generator.randomObjectName()).build();
+    byte[] bytes = DataGenerator.rand(rand).genBytes(1);
+
+    BlobWriteSession session = storage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+    try {
+      try (WritableByteChannel channel = session.open()) {
+        channel.write(ByteBuffer.wrap(bytes));
+      }
+      // it is okay if the exception is raised during write itself or close, if it happens during
+      // close we should get an AsynchronousCloseException
+    } catch (AsynchronousCloseException ace) {
+      assertThat(ace).hasCauseThat().hasMessageThat().contains("NOT_FOUND");
+      assertThat(((StorageException) ace.getCause()).getCode()).isEqualTo(404);
+    } catch (StorageException se) {
+      assertThat(se.getCode()).isEqualTo(404);
+    }
+
+    // the result future should resolve to a failure specifying the failure kind
+    StorageException se =
+        assertThrows(
+            StorageException.class,
+            () -> ApiExceptions.callAndTranslateApiException(session.getResult()));
+    assertThat(se.getCode()).isEqualTo(404);
+  }
+
+  @Test
+  public void uploadingAnObjectWorks() throws Exception {
+    doTest(bucket, 32 * _1MiB + 37, ImmutableList.of(), ImmutableList.of(), ImmutableList.of());
+  }
+
+  @Test
+  public void uploadingAnObjectWorks_requesterPays() throws Exception {
+    String projectId = storage.getOptions().getProjectId();
+    int _1MiB = 1024 * 1024;
+    doTest(
+        rpBucket,
+        32 * _1MiB + 37,
+        ImmutableList.of(BlobTargetOption.userProject(projectId)),
+        ImmutableList.of(BlobWriteOption.userProject(projectId)),
+        ImmutableList.of(BlobSourceOption.userProject(projectId)));
+  }
+
+  @Test
+  public void uploadingAnObjectWorks_customerSuppliedEncryptionKey() throws IOException {
+    CSEKSupport csek = CSEKSupport.create();
+    Key key = csek.getKey();
+
+    doTest(
+        bucket,
+        16 * _1MiB - 13,
+        ImmutableList.of(BlobTargetOption.encryptionKey(key)),
+        ImmutableList.of(BlobWriteOption.encryptionKey(key)),
+        ImmutableList.of(BlobSourceOption.decryptionKey(key)));
+  }
+
+  @Test
+  public void uploadingAnObjectWorks_kms() throws IOException {
+    CryptoKey key1 = kmsFixture.getKey1();
+    doTest(
+        bucket,
+        16 * _1MiB - 13,
+        ImmutableList.of(BlobTargetOption.kmsKeyName(key1.getName())),
+        ImmutableList.of(BlobWriteOption.kmsKeyName(key1.getName())),
+        ImmutableList.of());
+  }
+
+  /**
+   * Create an empty object, then overwrite it using a Parallel Composite Upload, then read the full
+   * object and verify its contents match exactly with what was written.
+   */
+  private void doTest(
+      BucketInfo bucket,
+      int objectSizeBytes,
+      ImmutableList<BlobTargetOption> overriddenCreateOptions,
+      ImmutableList<BlobWriteOption> overriddenOverwriteOptions,
+      ImmutableList<BlobSourceOption> overriddenReadOptions)
+      throws IOException {
+
+    BlobTargetOption[] createOptions =
+        BlobTargetOption.dedupe(
+            ImmutableList.of(BlobTargetOption.doesNotExist()),
+            overriddenCreateOptions.toArray(new BlobTargetOption[0]));
+    BlobWriteOption[] overwriteOptions =
+        BlobWriteOption.dedupe(
+            ImmutableList.of(BlobWriteOption.generationMatch()),
+            overriddenOverwriteOptions.toArray(new BlobWriteOption[0]));
+    BlobSourceOption[] readOptions =
+        BlobSourceOption.dedupe(
+            ImmutableList.of(), overriddenReadOptions.toArray(new BlobSourceOption[0]));
+
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    Blob gen1 = storage.create(info, createOptions);
+
+    byte[] bytes = DataGenerator.rand(rand).genBytes(objectSizeBytes);
+
+    BlobWriteSession session = storage.blobWriteSession(gen1, overwriteOptions);
+
+    try (WritableByteChannel channel = session.open()) {
+      long written = channel.write(ByteBuffer.wrap(bytes));
+      assertThat(written).isEqualTo(objectSizeBytes);
+    }
+
+    BlobInfo result = ApiExceptions.callAndTranslateApiException(session.getResult());
+
+    assertThat(result.getCrc32c()).isNotNull();
+    assertThat(result.getGeneration()).isNotNull();
+
+    byte[] actual = storage.readAllBytes(result.getBlobId(), readOptions);
+
+    assertThat(actual).isEqualTo(bytes);
+    assertThat(xxd(actual)).isEqualTo(xxd(bytes));
+  }
+}