Repository: googleapis/java-storage
Pull request #2230 (2023-09-26 23:26:05)

chore: add ParallelCompositeUploadWritableByteChannel
_Pre-Work_

Add new ParallelCompositeUploadWritableByteChannel which allows calls to `write` which will automatically break the stream into part objects, uploading each part in the background.

Error handling and validation is implemented with several tests (The majority of this PR is tests, as inducing some of the failures is rather involved).

Automatic cleanup in the case of failure will be implemented in a follow up pr.



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobId.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobId.java
index 6d058997..90dfdea6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobId.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobId.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.api.core.InternalApi;
 import com.google.common.base.MoreObjects;
 import java.io.Serializable;
 import java.util.Objects;
@@ -104,6 +105,11 @@ public final class BlobId implements Serializable {
         && Objects.equals(generation, other.generation);
   }
 
+  @InternalApi
+  BlobId withGeneration(long generation) {
+    return new BlobId(bucket, name, generation);
+  }
+
   /**
    * Creates a blob identifier. Generation is set to {@code null}.
    *
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
index e793bcfe..e72613fe 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
@@ -60,6 +60,11 @@ final class Buffers {
     return b.position();
   }
 
+  /** attempt to drain all of {@code content} into {@code dst} */
+  static long copy(ByteBuffer content, ByteBuffer dst) {
+    return copy(content, new ByteBuffer[] {dst}, 0, 1);
+  }
+
   /**
    * attempt to drain all of {@code content} into {@code dsts} starting from {@code dsts[0]} through
    * {@code dsts[dsts.length - 1]}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
index 45543f00..e66f04ea 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage;
 
+import java.util.Objects;
+
 abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
 
   private Crc32cValue() {}
@@ -95,6 +97,23 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
     public Crc32cLengthKnown withLength(long length) {
       return new Crc32cLengthKnown(value, length);
     }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Crc32cLengthUnknown)) {
+        return false;
+      }
+      Crc32cLengthUnknown that = (Crc32cLengthUnknown) o;
+      return value == that.value;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(value);
+    }
   }
 
   static final class Crc32cLengthKnown extends Crc32cValue<Crc32cLengthKnown> {
@@ -130,5 +149,22 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
     public String debugString() {
       return fmtCrc32cValue(value);
     }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Crc32cLengthKnown)) {
+        return false;
+      }
+      Crc32cLengthKnown that = (Crc32cLengthKnown) o;
+      return value == that.value && length == that.length;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(value, length);
+    }
   }
 }
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 19137ee2..c061cae8 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
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.ByteSizeConstants._16MiB;
+import static com.google.cloud.storage.ByteSizeConstants._1MiB;
 import static com.google.cloud.storage.ByteSizeConstants._256KiB;
 import static com.google.cloud.storage.CrossTransportUtils.fmtMethodName;
 import static com.google.cloud.storage.CrossTransportUtils.throwHttpJsonOnly;
@@ -149,6 +150,7 @@ import java.util.function.Predicate;
 import java.util.function.UnaryOperator;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
+import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 @BetaApi
@@ -620,7 +622,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     }
 
     if (copyRequest.getMegabytesCopiedPerChunk() != null) {
-      b.setMaxBytesRewrittenPerCall(copyRequest.getMegabytesCopiedPerChunk());
+      b.setMaxBytesRewrittenPerCall(copyRequest.getMegabytesCopiedPerChunk() * _1MiB);
     }
 
     RewriteObjectRequest req = mapper.apply(b).build();
@@ -1968,30 +1970,44 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
         Decoder.identity());
   }
 
-  @Nullable
-  private Blob internalBlobGet(BlobId blob, Opts<ObjectSourceOpt> unwrap) {
-    Opts<ObjectSourceOpt> opts =
-        unwrap.resolveFrom(blob).prepend(defaultOpts).prepend(ALL_BLOB_FIELDS);
+  @NonNull
+  @Override
+  public BlobInfo internalObjectGet(BlobId blobId, Opts<ObjectSourceOpt> opts) {
+    Opts<ObjectSourceOpt> finalOpts = opts.prepend(defaultOpts).prepend(ALL_BLOB_FIELDS);
     GrpcCallContext grpcCallContext =
-        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+        finalOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     GetObjectRequest.Builder builder =
         GetObjectRequest.newBuilder()
-            .setBucket(bucketNameCodec.encode(blob.getBucket()))
-            .setObject(blob.getName());
-    ifNonNull(blob.getGeneration(), builder::setGeneration);
-    GetObjectRequest req = opts.getObjectsRequest().apply(builder).build();
+            .setBucket(bucketNameCodec.encode(blobId.getBucket()))
+            .setObject(blobId.getName());
+    ifNonNull(blobId.getGeneration(), builder::setGeneration);
+    GetObjectRequest req = finalOpts.getObjectsRequest().apply(builder).build();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
+    //noinspection DataFlowIssue
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
-        () -> {
-          try {
-            return storageClient.getObjectCallable().call(req, merge);
-          } catch (NotFoundException ignore) {
-            return null;
-          }
-        },
-        syntaxDecoders.blob.andThen(opts.clearBlobFields()));
+        () -> storageClient.getObjectCallable().call(req, merge),
+        resp -> {
+          BlobInfo tmp = codecs.blobInfo().decode(resp);
+          return finalOpts.clearBlobFields().decode(tmp);
+        });
+  }
+
+  @Nullable
+  private Blob internalBlobGet(BlobId blob, Opts<ObjectSourceOpt> unwrap) {
+    Opts<ObjectSourceOpt> opts = unwrap.resolveFrom(blob);
+    try {
+      return internalObjectGet(blob, opts).asBlob(this);
+    } catch (StorageException e) {
+      if (e.getCause() instanceof NotFoundException) {
+        return null;
+      } else {
+        throw e;
+      }
+    } catch (NotFoundException nfe) {
+      return null;
+    }
   }
 
   @Nullable
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadException.java
new file mode 100644
index 00000000..b03348e5
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadException.java
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ErrorDetails;
+import com.google.api.gax.rpc.StatusCode;
+import com.google.common.collect.ImmutableList;
+import io.grpc.Status.Code;
+
+final class ParallelCompositeUploadException extends ApiException {
+
+  private final ApiFuture<ImmutableList<BlobId>> createdObjects;
+
+  private ParallelCompositeUploadException(
+      Throwable cause,
+      StatusCode statusCode,
+      ErrorDetails errorDetails,
+      ApiFuture<ImmutableList<BlobId>> createdObjects) {
+    super(cause, statusCode, false, errorDetails);
+    this.createdObjects = createdObjects;
+  }
+
+  public ApiFuture<ImmutableList<BlobId>> getCreatedObjects() {
+    return createdObjects;
+  }
+
+  static ParallelCompositeUploadException of(
+      Throwable t, ApiFuture<ImmutableList<BlobId>> createdObjects) {
+    StatusCode statusCode;
+    ErrorDetails errorDetails;
+
+    Throwable cause = t;
+    if (t instanceof StorageException && t.getCause() != null) {
+      cause = t.getCause();
+    }
+
+    if (cause instanceof ApiException) {
+      ApiException apiException = (ApiException) cause;
+      statusCode = apiException.getStatusCode();
+      errorDetails = apiException.getErrorDetails();
+    } else {
+      statusCode = GrpcStatusCode.of(Code.UNKNOWN);
+      errorDetails = ErrorDetails.builder().build();
+    }
+    return new ParallelCompositeUploadException(cause, statusCode, errorDetails, createdObjects);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
new file mode 100644
index 00000000..72473b32
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
@@ -0,0 +1,496 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutureCallback;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.ApiFutureUtils.OnFailureApiFutureCallback;
+import com.google.cloud.storage.ApiFutureUtils.OnSuccessApiFutureCallback;
+import com.google.cloud.storage.AsyncAppendingQueue.ShortCircuitException;
+import com.google.cloud.storage.BufferHandlePool.PooledBuffer;
+import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
+import com.google.cloud.storage.MetadataField.PartRange;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.UnifiedOpts.Crc32cMatch;
+import com.google.cloud.storage.UnifiedOpts.GenerationMatch;
+import com.google.cloud.storage.UnifiedOpts.GenerationNotMatch;
+import com.google.cloud.storage.UnifiedOpts.Md5Match;
+import com.google.cloud.storage.UnifiedOpts.MetagenerationMatch;
+import com.google.cloud.storage.UnifiedOpts.MetagenerationNotMatch;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.UnifiedOpts.SourceGenerationMatch;
+import com.google.cloud.storage.UnifiedOpts.SourceGenerationNotMatch;
+import com.google.cloud.storage.UnifiedOpts.SourceMetagenerationMatch;
+import com.google.cloud.storage.UnifiedOpts.SourceMetagenerationNotMatch;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+import io.grpc.Status.Code;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.Executor;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@SuppressWarnings("UnstableApiUsage") // guava hashing
+final class ParallelCompositeUploadWritableByteChannel implements BufferedWritableByteChannel {
+
+  private static final MetadataField<String> FINAL_OBJECT_NAME =
+      MetadataField.forString("pcu_finalObjectName");
+  private static final MetadataField<PartRange> PART_INDEX =
+      MetadataField.forPartRange("pcu_partIndex");
+  private static final MetadataField<Long> OBJECT_OFFSET =
+      MetadataField.forLong("pcu_objectOffset");
+  private static final Comparator<BlobInfo> comparator =
+      Comparator.comparing(PART_INDEX::readFrom, PartRange.COMP);
+  private static final Predicate<ObjectTargetOpt> TO_EXCLUDE_FROM_PARTS;
+  // when creating a part or composing we include a precondition so that it can be retried
+  private static final Opts<GenerationMatch> DOES_NOT_EXIST =
+      Opts.from(UnifiedOpts.generationMatch(0));
+
+  static {
+    //noinspection deprecation
+    Predicate<ObjectTargetOpt> tmp =
+        o ->
+            o instanceof GenerationMatch
+                || o instanceof GenerationNotMatch
+                || o instanceof MetagenerationMatch
+                || o instanceof MetagenerationNotMatch
+                || o instanceof SourceGenerationMatch
+                || o instanceof SourceGenerationNotMatch
+                || o instanceof SourceMetagenerationMatch
+                || o instanceof SourceMetagenerationNotMatch
+                || o instanceof Crc32cMatch
+                || o instanceof Md5Match;
+    TO_EXCLUDE_FROM_PARTS = tmp.negate();
+  }
+
+  // immutable provided values
+  private final BufferHandlePool bufferPool;
+  private final Executor exec;
+  private final PartNamingStrategy partNamingStrategy;
+  private final PartCleanupStrategy partCleanupStrategy;
+  private final int maxElementsPerCompact;
+  private final SettableApiFuture<BlobInfo> finalObject;
+  private final StorageInternal storage;
+  private final BlobInfo ultimateObject;
+  private final Opts<ObjectTargetOpt> opts;
+
+  // immutable bootstrapped state
+  private final Opts<ObjectTargetOpt> partOpts;
+  private final AsyncAppendingQueue<BlobInfo> queue;
+  private final FailureForwarder failureForwarder;
+  // mutable running state
+  private final List<ApiFuture<BlobInfo>> pendingParts;
+  private final List<BlobId> successfulParts;
+  private final Hasher cumulativeHasher;
+  private boolean open;
+  private long totalObjectOffset;
+  private PooledBuffer current;
+
+  ParallelCompositeUploadWritableByteChannel(
+      BufferHandlePool bufferPool,
+      Executor exec,
+      PartNamingStrategy partNamingStrategy,
+      PartCleanupStrategy partCleanupStrategy,
+      int maxElementsPerCompact,
+      SettableApiFuture<BlobInfo> finalObject,
+      StorageInternal storage,
+      BlobInfo ultimateObject,
+      Opts<ObjectTargetOpt> opts) {
+    this.bufferPool = bufferPool;
+    this.exec = exec;
+    this.partNamingStrategy = partNamingStrategy;
+    this.partCleanupStrategy = partCleanupStrategy;
+    this.maxElementsPerCompact = maxElementsPerCompact;
+    this.finalObject = finalObject;
+    this.storage = storage;
+    this.ultimateObject = ultimateObject;
+    this.opts = opts;
+    this.queue = AsyncAppendingQueue.of(exec, maxElementsPerCompact, this::compose);
+    this.pendingParts = new ArrayList<>();
+    // this can be modified by another thread
+    this.successfulParts = Collections.synchronizedList(new ArrayList<>());
+    this.open = true;
+    this.totalObjectOffset = 0;
+
+    this.partOpts = getPartOpts(opts);
+    this.cumulativeHasher = Hashing.crc32c().newHasher();
+    this.failureForwarder = new FailureForwarder();
+  }
+
+  @Override
+  public synchronized int write(ByteBuffer src) throws IOException {
+    if (!open) {
+      throw new ClosedChannelException();
+    }
+
+    int remaining = src.remaining();
+    cumulativeHasher.putBytes(src.duplicate());
+    while (src.hasRemaining()) {
+      if (current == null) {
+        current = bufferPool.getBuffer();
+      }
+
+      ByteBuffer buf = current.getBufferHandle().get();
+      Buffers.copy(src, buf);
+
+      if (!buf.hasRemaining()) {
+        internalFlush(buf);
+      }
+    }
+
+    return remaining;
+  }
+
+  @Override
+  public synchronized boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public synchronized void flush() throws IOException {
+    if (current != null) {
+      ByteBuffer buf = current.getBufferHandle().get();
+      internalFlush(buf);
+    }
+  }
+
+  @Override
+  public synchronized void close() throws IOException {
+    if (!open) {
+      return;
+    }
+    open = false;
+
+    flush();
+
+    try {
+      queue.close();
+    } catch (NoSuchElementException e) {
+      // We never created any parts
+      // create an empty object
+      try {
+        BlobInfo blobInfo = storage.internalDirectUpload(ultimateObject, opts, Buffers.allocate(0));
+        finalObject.set(blobInfo);
+        return;
+      } catch (StorageException se) {
+        finalObject.setException(se);
+        throw se;
+      }
+    }
+
+    String expectedCrc32c = Utils.crc32cCodec.encode(cumulativeHasher.hash().asInt());
+    ApiFuture<BlobInfo> closingTransform =
+        ApiFutures.transformAsync(queue.getResult(), this::cleanupParts, exec);
+    ApiFuture<BlobInfo> validatingTransform =
+        ApiFutures.transformAsync(
+            closingTransform,
+            finalInfo -> {
+              String crc32c = finalInfo.getCrc32c();
+              if (expectedCrc32c.equals(crc32c)) {
+                return ApiFutures.immediateFuture(finalInfo);
+              } else {
+                return ApiFutures.immediateFailedFuture(
+                    StorageException.coalesce(
+                        buildParallelCompositeUploadException(
+                            ApiExceptionFactory.createException(
+                                String.format(
+                                    "CRC32C Checksum mismatch. expected: [%s] but was: [%s]",
+                                    expectedCrc32c, crc32c),
+                                null,
+                                GrpcStatusCode.of(Code.DATA_LOSS),
+                                false),
+                            exec,
+                            pendingParts,
+                            successfulParts)));
+              }
+            },
+            exec);
+
+    if (partCleanupStrategy.isDeleteOnError()) {
+      ApiFuture<BlobInfo> cleaningFuture =
+          ApiFutures.catchingAsync(
+              validatingTransform,
+              Throwable.class,
+              t -> {
+                // todo:
+                return ApiFutures.immediateFailedFuture(t);
+              },
+              exec);
+      // todo: verify this gets the first failure and not one from cleaning
+      ApiFutures.addCallback(cleaningFuture, failureForwarder, exec);
+    } else {
+      ApiFutures.addCallback(validatingTransform, failureForwarder, exec);
+    }
+
+    // we don't need the value from this, but we do need any exception that might be present
+    try {
+      ApiFutureUtils.await(validatingTransform);
+    } catch (Throwable t) {
+      AsynchronousCloseException e = new AsynchronousCloseException();
+      e.initCause(t);
+      throw e;
+    }
+  }
+
+  private void internalFlush(ByteBuffer buf) {
+    Buffers.flip(buf);
+    int pendingByteCount = buf.remaining();
+    int partIndex = pendingParts.size() + 1;
+    BlobInfo partInfo = definePart(ultimateObject, PartRange.of(partIndex), totalObjectOffset);
+    ApiFuture<BlobInfo> partFuture =
+        ApiFutures.transform(
+            ApiFutures.immediateFuture(partInfo),
+            info -> {
+              try {
+                return storage.internalDirectUpload(info, partOpts, buf);
+              } catch (StorageException e) {
+                // a precondition failure usually means the part was created, but we didn't get the
+                // response. And when we tried to retry the object already exists.
+                if (e.getCode() == 412) {
+                  Opts<ObjectSourceOpt> srcOpts = partOpts.transformTo(ObjectSourceOpt.class);
+                  return storage.internalObjectGet(info.getBlobId(), srcOpts);
+                } else {
+                  throw e;
+                }
+              }
+            },
+            exec);
+
+    ApiFutures.addCallback(
+        partFuture,
+        new BufferHandleReleaser<>(
+            bufferPool,
+            current,
+            (OnSuccessApiFutureCallback<BlobInfo>)
+                result -> successfulParts.add(result.getBlobId())),
+        exec);
+
+    pendingParts.add(partFuture);
+    try {
+      queue.append(partFuture);
+      totalObjectOffset += pendingByteCount;
+    } catch (ShortCircuitException e) {
+      open = false;
+      bufferPool.returnBuffer(current);
+
+      // attempt to cancel any pending requests which haven't started yet
+      for (ApiFuture<BlobInfo> pendingPart : pendingParts) {
+        pendingPart.cancel(false);
+      }
+
+      Throwable cause = e.getCause();
+      // create our exception containing information about the upload context
+      ParallelCompositeUploadException pcue =
+          buildParallelCompositeUploadException(cause, exec, pendingParts, successfulParts);
+      BaseServiceException storageException = StorageException.coalesce(pcue);
+
+      // asynchronously fail the finalObject future
+      CancellationException cancellationException =
+          new CancellationException(storageException.getMessage());
+      cancellationException.initCause(storageException);
+      ApiFutures.addCallback(
+          ApiFutures.immediateFailedFuture(cancellationException),
+          (OnFailureApiFutureCallback<BlobInfo>) failureForwarder::onFailure,
+          exec);
+      if (partCleanupStrategy.isDeleteOnError()) {
+        // TODO: cleanup
+      }
+      throw storageException;
+    } finally {
+      current = null;
+    }
+  }
+
+  @SuppressWarnings("DataFlowIssue")
+  private BlobInfo compose(ImmutableList<BlobInfo> parts) {
+    ComposeRequest.Builder builder = ComposeRequest.newBuilder();
+
+    List<BlobInfo> sorted = parts.stream().sorted(comparator).collect(Collectors.toList());
+
+    sorted.stream()
+        .map(BlobInfo::getBlobId)
+        .forEach(id -> builder.addSource(id.getName(), id.getGeneration()));
+
+    if (parts.size() == maxElementsPerCompact) {
+      // perform an intermediary compose
+      BlobInfo first = sorted.get(0);
+      BlobInfo last = sorted.get(sorted.size() - 1);
+
+      long firstIdx = PART_INDEX.readFrom(first).getBegin();
+      long lastIdx = PART_INDEX.readFrom(last).getEnd();
+      long offset = OBJECT_OFFSET.readFrom(first);
+      BlobInfo newPart = definePart(ultimateObject, PartRange.of(firstIdx, lastIdx), offset);
+      builder.setTarget(newPart);
+      builder.setTargetOpts(partOpts);
+    } else {
+      // with this compose create the ultimate object
+      builder.setTarget(ultimateObject);
+      builder.setTargetOpts(opts);
+    }
+
+    ComposeRequest composeRequest = builder.build();
+    BlobInfo compose = storage.compose(composeRequest);
+    successfulParts.add(compose.getBlobId());
+    return compose;
+  }
+
+  private ApiFuture<BlobInfo> cleanupParts(BlobInfo finalInfo) {
+    if (!partCleanupStrategy.isDeleteParts()) {
+      return ApiFutures.immediateFuture(finalInfo);
+    }
+    List<ApiFuture<Boolean>> deletes =
+        successfulParts.stream()
+            // make sure we don't delete the object we're wanting to create
+            .filter(id -> !id.equals(finalInfo.getBlobId()))
+            .map(ApiFutures::immediateFuture)
+            .map(f -> ApiFutures.transform(f, storage::delete, exec))
+            .collect(Collectors.toList());
+
+    ApiFuture<List<Boolean>> deletes2 = ApiFutureUtils.quietAllAsList(deletes);
+
+    return ApiFutures.transform(deletes2, ignore -> finalInfo, exec);
+  }
+
+  private BlobInfo definePart(BlobInfo ultimateObject, PartRange partRange, long offset) {
+    BlobId id = ultimateObject.getBlobId();
+    BlobInfo.Builder b = ultimateObject.toBuilder();
+    String partName = partNamingStrategy.fmtName(id.getName(), partRange);
+    b.setBlobId(BlobId.of(id.getBucket(), partName));
+    ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+    Map<@NonNull String, @Nullable String> metadata = ultimateObject.getMetadata();
+    if (metadata != null) {
+      builder.putAll(metadata);
+    }
+    FINAL_OBJECT_NAME.appendTo(id.getName(), builder);
+    PART_INDEX.appendTo(partRange, builder);
+    OBJECT_OFFSET.appendTo(offset, builder);
+    b.setMetadata(builder.build());
+    return b.build();
+  }
+
+  @VisibleForTesting
+  @NonNull
+  static Opts<ObjectTargetOpt> getPartOpts(Opts<ObjectTargetOpt> opts) {
+    return opts.filter(TO_EXCLUDE_FROM_PARTS).prepend(DOES_NOT_EXIST);
+  }
+
+  @VisibleForTesting
+  static final class BufferHandleReleaser<T> implements ApiFutureCallback<T> {
+    private final BufferHandlePool bufferManager;
+    private final ApiFutureCallback<T> delegate;
+    private final PooledBuffer toRelease;
+
+    @VisibleForTesting
+    BufferHandleReleaser(
+        BufferHandlePool bufferPool, PooledBuffer toRelease, ApiFutureCallback<T> delegate) {
+      this.bufferManager = bufferPool;
+      this.delegate = delegate;
+      this.toRelease = toRelease;
+    }
+
+    @Override
+    public void onFailure(Throwable t) {
+      try {
+        delegate.onFailure(t);
+      } finally {
+        bufferManager.returnBuffer(toRelease);
+      }
+    }
+
+    @Override
+    public void onSuccess(T result) {
+      try {
+        delegate.onSuccess(result);
+      } finally {
+        bufferManager.returnBuffer(toRelease);
+      }
+    }
+  }
+
+  private class FailureForwarder implements ApiFutureCallback<BlobInfo> {
+
+    @Override
+    public void onFailure(Throwable t) {
+      finalObject.setException(t);
+    }
+
+    @Override
+    public void onSuccess(BlobInfo result) {
+      finalObject.set(result);
+    }
+  }
+
+  @VisibleForTesting
+  @NonNull
+  static ParallelCompositeUploadException buildParallelCompositeUploadException(
+      Throwable cause,
+      Executor exec,
+      List<ApiFuture<BlobInfo>> pendingParts,
+      List<BlobId> successfulParts) {
+    ApiFuture<List<BlobInfo>> successfulList = ApiFutures.successfulAsList(pendingParts);
+    // suppress any failure that might happen when waiting for any pending futures to resolve
+    ApiFuture<List<BlobInfo>> catching =
+        ApiFutures.catching(successfulList, Throwable.class, t2 -> ImmutableList.of(), exec);
+
+    ApiFuture<ImmutableList<BlobId>> fCreatedObjects =
+        ApiFutures.transform(
+            catching,
+            l ->
+                Stream.of(
+                        l.stream()
+                            // any failed future from successfulList will contain a null value
+                            // filter out as that means an object wasn't created
+                            .filter(Objects::nonNull)
+                            .map(BlobInfo::getBlobId),
+                        successfulParts.stream())
+                    .flatMap(Function.identity()) // .flatten()
+                    // a successful value could be in successfulParts and pendingParts if
+                    // pendingParts haven't compacted yet
+                    .distinct()
+                    .collect(ImmutableList.toImmutableList()),
+            exec);
+
+    return ParallelCompositeUploadException.of(cause, fCreatedObjects);
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
index 9674eb28..6249b0d4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage;
 
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import java.io.IOException;
@@ -40,4 +42,12 @@ interface StorageInternal {
   default boolean delete(BlobId id) {
     throw new UnsupportedOperationException("not implemented");
   }
+
+  default BlobInfo compose(ComposeRequest composeRequest) {
+    throw new UnsupportedOperationException("not implemented");
+  }
+
+  default BlobInfo internalObjectGet(BlobId blobId, Opts<ObjectSourceOpt> opts) {
+    throw new UnsupportedOperationException("not implemented");
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index bbafc35f..9e0be115 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -127,7 +127,7 @@ final class UnifiedOpts {
   private interface ApplicableHmacKey {}
 
   /** Base interface for those Opts which may expose their values via gRPC Metadata */
-  private interface GrpcMetadataMapper {
+  private interface GrpcMetadataMapper extends Opt {
     default Mapper<GrpcCallContext> getGrpcMetadataMapper() {
       return Mapper.identity();
     }
@@ -646,13 +646,19 @@ final class UnifiedOpts {
   }
 
   /** @see DecryptionKey */
-  static final class EncryptionKey extends RpcOptVal<Key> implements ObjectTargetOpt {
+  static final class EncryptionKey extends RpcOptVal<Key>
+      implements ObjectTargetOpt, ProjectAsSource<DecryptionKey> {
     private static final long serialVersionUID = -7335988656032764620L;
 
     private EncryptionKey(Key val) {
       super(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, val);
     }
 
+    @Override
+    public DecryptionKey asSource() {
+      return new DecryptionKey(val);
+    }
+
     @Override
     public Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> mapper() {
       return b ->
@@ -798,13 +804,13 @@ final class UnifiedOpts {
      * However, refactoring the whole model pipeline for both json and grpc is going to be a large
      * change.
      */
-    Decoder<Blob, Blob> clearUnselectedBlobFields() {
+    Decoder<BlobInfo, BlobInfo> clearUnselectedBlobFields() {
       return b -> {
         if (val.isEmpty()) {
           return b;
         } else {
           Set<String> names = getPaths();
-          Blob.Builder bldr = b.toBuilder();
+          BlobInfo.Builder bldr = b.toBuilder();
           blobInfoFieldClearers.entrySet().stream()
               .filter(e -> !names.contains(e.getKey()))
               .map(Entry::getValue)
@@ -2307,7 +2313,7 @@ final class UnifiedOpts {
       return filterTo(ReturnRawInputStream.class).findFirst().map(r -> r.val).orElse(true);
     }
 
-    Decoder<Blob, Blob> clearBlobFields() {
+    Decoder<BlobInfo, BlobInfo> clearBlobFields() {
       return filterTo(Fields.class)
           .findFirst()
           .map(Fields::clearUnselectedBlobFields)
@@ -2352,6 +2358,32 @@ final class UnifiedOpts {
       return new Opts<>(filterTo(c).collect(ImmutableList.toImmutableList()));
     }
 
+    Opts<T> filter(Predicate<T> p) {
+      return new Opts<>(opts.stream().filter(p).collect(ImmutableList.toImmutableList()));
+    }
+
+    <R extends Opt> Opts<R> transformTo(Class<R> c) {
+      return new Opts<>(
+          opts.stream()
+              .map(
+                  o -> {
+                    if (o instanceof ProjectAsSource) {
+                      ProjectAsSource<?> projectAsSource = (ProjectAsSource<?>) o;
+                      Opt asSource = projectAsSource.asSource();
+                      if (c.isAssignableFrom(asSource.getClass())) {
+                        return c.cast(asSource);
+                      }
+                    }
+                    if (c.isAssignableFrom(o.getClass())) {
+                      return c.cast(o);
+                    } else {
+                      return null;
+                    }
+                  })
+              .filter(Objects::nonNull)
+              .collect(ImmutableList.toImmutableList()));
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) {
@@ -2373,12 +2405,12 @@ final class UnifiedOpts {
       return fuseMappers(RpcOptVal.class, RpcOptVal::mapper);
     }
 
-    private <R, O> Mapper<O> fuseMappers(Class<R> c, Function<R, Mapper<O>> f) {
+    private <R extends Opt, O> Mapper<O> fuseMappers(Class<R> c, Function<R, Mapper<O>> f) {
       return filterTo(c).map(f).reduce(Mapper.identity(), Mapper::andThen);
     }
 
     @SuppressWarnings("unchecked")
-    private <R> Stream<R> filterTo(Class<R> c) {
+    private <R extends Opt> Stream<R> filterTo(Class<R> c) {
       // TODO: figure out if there is need for an "isApplicableTo" predicate
       return opts.stream().filter(isInstanceOf(c)).map(x -> (R) x);
     }
@@ -2416,7 +2448,7 @@ final class UnifiedOpts {
     }
 
     /** Create a predicate which is able to effectively perform an {@code instanceof} check */
-    private static <T> Predicate<T> isInstanceOf(Class<?> c) {
+    private static <T extends Opt, O extends Opt> Predicate<T> isInstanceOf(Class<O> c) {
       return t -> c.isAssignableFrom(t.getClass());
     }
   }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannelTest.java
new file mode 100644
index 00000000..ec21db9a
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannelTest.java
@@ -0,0 +1,895 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.core.ApiFutureCallback;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.cloud.storage.BufferHandlePool.PooledBuffer;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.MetadataField.PartRange;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadWritableByteChannel.BufferHandleReleaser;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.Storage.PredefinedAcl;
+import com.google.cloud.storage.UnifiedOpts.EncryptionKey;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.common.util.concurrent.Uninterruptibles;
+import com.google.storage.v2.WriteObjectRequest;
+import io.grpc.Status.Code;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.junit.Before;
+import org.junit.Test;
+
+public final class ParallelCompositeUploadWritableByteChannelTest {
+
+  private static final Hasher HASHER = Hasher.enabled();
+
+  private BlobInfo info;
+  private Opts<ObjectTargetOpt> opts;
+
+  private BufferHandlePool bufferHandlePool;
+  private SettableApiFuture<BlobInfo> finalObject;
+  private FakeStorageInternal storageInternal;
+  private SimplisticPartNamingStrategy partNamingStrategy;
+  private int bufferCapacity;
+
+  @Before
+  public void setUp() throws Exception {
+    info = BlobInfo.newBuilder("bucket", "object").build();
+    opts = Opts.from(UnifiedOpts.generationMatch(0));
+    bufferCapacity = 10;
+    bufferHandlePool = BufferHandlePool.simple(bufferCapacity);
+    finalObject = SettableApiFuture.create();
+    partNamingStrategy = new SimplisticPartNamingStrategy("prefix");
+    storageInternal = new FakeStorageInternal();
+  }
+
+  @Test
+  public void objectCreated_partCount_eqToLimit() throws Exception {
+
+    int maxElementsPerCompact = 5;
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(maxElementsPerCompact);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(47);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+
+    String name = info.getName();
+    // individual parts
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    BlobId p3 = id(partNamingStrategy.fmtName(name, PartRange.of(3)), 3L);
+    BlobId p4 = id(partNamingStrategy.fmtName(name, PartRange.of(4)), 4L);
+    BlobId p5 = id(partNamingStrategy.fmtName(name, PartRange.of(5)), 5L);
+    // compose
+    BlobId c1 = id(partNamingStrategy.fmtName(name, PartRange.of(1, 5)), 6L);
+    // ultimate object
+    BlobId expectedId = id(name, 7L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () ->
+            assertThat(storageInternal.addedObjects.keySet())
+                .containsExactly(p1, p2, p3, p4, p5, c1, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).containsExactly(p1, p2, p3, p4, p5, c1));
+  }
+
+  @Test
+  public void objectCreated_partCount_ltToLimit() throws Exception {
+
+    int maxElementsPerCompact = 6;
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(maxElementsPerCompact);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(47);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+
+    String name = info.getName();
+    // individual parts
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    BlobId p3 = id(partNamingStrategy.fmtName(name, PartRange.of(3)), 3L);
+    BlobId p4 = id(partNamingStrategy.fmtName(name, PartRange.of(4)), 4L);
+    BlobId p5 = id(partNamingStrategy.fmtName(name, PartRange.of(5)), 5L);
+    // no compose
+    // ultimate object
+    BlobId expectedId = id(name, 6L);
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () ->
+            assertThat(storageInternal.addedObjects.keySet())
+                .containsExactly(p1, p2, p3, p4, p5, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).containsExactly(p1, p2, p3, p4, p5));
+  }
+
+  @Test
+  public void objectCreated_partCount_gtToLimit() throws Exception {
+
+    int maxElementsPerCompact = 4;
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(maxElementsPerCompact);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(47);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+
+    String name = info.getName();
+    // parts 1-4
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    BlobId p3 = id(partNamingStrategy.fmtName(name, PartRange.of(3)), 3L);
+    BlobId p4 = id(partNamingStrategy.fmtName(name, PartRange.of(4)), 4L);
+    // compose 1-4
+    BlobId c1 = id(partNamingStrategy.fmtName(name, PartRange.of(1, 4)), 5L);
+    // part 5
+    BlobId p5 = id(partNamingStrategy.fmtName(name, PartRange.of(5)), 6L);
+    // ultimate object
+    BlobId expectedId = id(name, 7L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () ->
+            assertThat(storageInternal.addedObjects.keySet())
+                .containsExactly(p1, p2, p3, p4, c1, p5, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).containsExactly(p1, p2, p3, p4, p5, c1));
+  }
+
+  @Test
+  public void cleanup_success_disabled() throws Exception {
+
+    int maxElementsPerCompact = 10;
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.never(),
+            maxElementsPerCompact,
+            finalObject,
+            storageInternal,
+            info,
+            opts);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(22);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+
+    String name = info.getName();
+    // parts 1-4
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    BlobId p3 = id(partNamingStrategy.fmtName(name, PartRange.of(3)), 3L);
+    // ultimate object
+    BlobId expectedId = id(name, 4L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () ->
+            assertThat(storageInternal.addedObjects.keySet())
+                .containsExactly(p1, p2, p3, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).isEmpty());
+  }
+
+  @Test
+  public void writeDoesNotFlushIfItIsnNotFull() throws Exception {
+
+    int maxElementsPerCompact = 10;
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.never(),
+            maxElementsPerCompact,
+            finalObject,
+            storageInternal,
+            info,
+            opts);
+
+    byte[] bytes1 = DataGenerator.base64Characters().genBytes(bufferCapacity * 2 - 1);
+    int limit = bufferCapacity - 1;
+    pcu.write(ByteBuffer.wrap(bytes1, 0, limit));
+    pcu.write(ByteBuffer.wrap(bytes1, limit, bytes1.length - limit));
+
+    pcu.close();
+
+    String name = info.getName();
+    // parts 1-4
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    // ultimate object
+    BlobId expectedId = id(name, 3L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () -> assertThat(storageInternal.addedObjects.keySet()).containsExactly(p1, p2, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).isEmpty(),
+        () -> {
+          Crc32cLengthKnown part1Crc32c = storageInternal.addedObjects.get(p1).getCrc32c();
+          Crc32cLengthKnown expected = HASHER.hash(ByteBuffer.wrap(bytes1, 0, bufferCapacity));
+          assertThat(part1Crc32c).isEqualTo(expected);
+        });
+  }
+
+  @Test
+  public void partOpts_stripsPreconditionsAndChecksums_addingIfGenEq0() {
+    EncryptionKey key = UnifiedOpts.encryptionKey("key");
+    Opts<ObjectTargetOpt> opts =
+        Opts.from(
+            UnifiedOpts.generationMatch(4),
+            UnifiedOpts.generationNotMatch(5),
+            UnifiedOpts.metagenerationMatch(6),
+            UnifiedOpts.metagenerationNotMatch(7),
+            UnifiedOpts.userProject("user-project"),
+            key,
+            UnifiedOpts.predefinedAcl(PredefinedAcl.PRIVATE),
+            UnifiedOpts.kmsKeyName("kms-key"),
+            UnifiedOpts.crc32cMatch(1),
+            UnifiedOpts.md5Match("asdf"),
+            UnifiedOpts.generationMatch(8).asSource(),
+            UnifiedOpts.generationNotMatch(10).asSource(),
+            UnifiedOpts.metagenerationMatch(12).asSource(),
+            UnifiedOpts.metagenerationNotMatch(14).asSource());
+
+    Opts<ObjectTargetOpt> partOpts = ParallelCompositeUploadWritableByteChannel.getPartOpts(opts);
+
+    ImmutableMap<StorageRpc.Option, ?> expected =
+        ImmutableMap.of(
+            StorageRpc.Option.IF_GENERATION_MATCH,
+            0L,
+            StorageRpc.Option.USER_PROJECT,
+            "user-project",
+            StorageRpc.Option.CUSTOMER_SUPPLIED_KEY,
+            Base64.getEncoder().encodeToString(key.val.getEncoded()),
+            StorageRpc.Option.PREDEFINED_ACL,
+            PredefinedAcl.PRIVATE.getEntry(),
+            StorageRpc.Option.KMS_KEY_NAME,
+            "kms-key");
+    ImmutableMap<StorageRpc.Option, ?> rpcOptions = partOpts.getRpcOptions();
+
+    assertThat(rpcOptions).isEqualTo(expected);
+  }
+
+  @Test
+  public void callingCloseOnANeverWrittenPcuStillCreatesAnEmptyObject() throws Exception {
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(3);
+    // never call write
+    pcu.close();
+    // ultimate object
+    BlobId expectedId = id(info.getName(), 1L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () -> assertThat(storageInternal.addedObjects.keySet()).containsExactly(expectedId),
+        () -> assertThat(storageInternal.deleteRequests).isEmpty());
+  }
+
+  @Test
+  public void partsRetainMetadata() throws Exception {
+    ImmutableMap<String, String> metadata = ImmutableMap.of("a", "1", "b", "2");
+
+    List<Map<String, String>> metadatas = Collections.synchronizedList(new ArrayList<>());
+    BlobInfo info = this.info.toBuilder().setMetadata(metadata).build();
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.never(),
+            3,
+            finalObject,
+            new FakeStorageInternal() {
+              @Override
+              public BlobInfo internalDirectUpload(
+                  BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+                metadatas.add(info.getMetadata());
+                return super.internalDirectUpload(info, opts, buf);
+              }
+
+              @Override
+              public BlobInfo compose(ComposeRequest composeRequest) {
+                metadatas.add(composeRequest.getTarget().getMetadata());
+                return super.compose(composeRequest);
+              }
+            },
+            info,
+            opts);
+
+    pcu.write(DataGenerator.base64Characters().genByteBuffer(bufferCapacity * 3 + 5));
+    pcu.close();
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+    assertAll(
+        () -> assertThat(result.getMetadata()).isEqualTo(metadata),
+        () -> {
+          assertThat(metadatas).isNotEmpty();
+          for (Map<String, String> m : metadatas) {
+            assertThat(m).containsAtLeastEntriesIn(metadata);
+          }
+        });
+  }
+
+  @Test
+  public void channelClosedException_writeAfterClose() throws Exception {
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(3);
+    // never call write
+    pcu.close();
+    assertThrows(
+        ClosedChannelException.class,
+        () -> pcu.write(DataGenerator.base64Characters().genByteBuffer(3)));
+  }
+
+  @Test
+  public void multipleInvocationsOfCloseDoNotError() throws Exception {
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(3);
+    pcu.close();
+    pcu.close();
+  }
+
+  @Test
+  public void openUponConstruction() throws Exception {
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(3);
+    assertThat(pcu.isOpen()).isTrue();
+    pcu.close();
+  }
+
+  @Test
+  public void callingFlushWhileBufferIsEmptyIsANoOp() throws Exception {
+    ParallelCompositeUploadWritableByteChannel pcu = defaultPcu(3);
+    pcu.write(DataGenerator.base64Characters().genByteBuffer(bufferCapacity));
+    pcu.flush();
+    pcu.close();
+
+    String name = info.getName();
+    // parts 1
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    // ultimate object
+    BlobId expectedId = id(name, 2L);
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () -> assertThat(storageInternal.addedObjects.keySet()).containsExactly(p1, expectedId),
+        () -> assertThat(storageInternal.composeRequests).hasSize(1),
+        () -> assertThat(storageInternal.deleteRequests).containsExactly(p1));
+  }
+
+  @Test
+  public void creatingAnEmptyObjectWhichFailsIsSetAsResultFailureAndThrowFromClose()
+      throws Exception {
+    //noinspection resource
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.always(),
+            3,
+            finalObject,
+            new FakeStorageInternal() {
+              @Override
+              public BlobInfo internalDirectUpload(
+                  BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+                throw StorageException.coalesce(
+                    ApiExceptionFactory.createException(
+                        null, GrpcStatusCode.of(Code.PERMISSION_DENIED), false));
+              }
+            },
+            info,
+            opts);
+    StorageException se1 = assertThrows(StorageException.class, pcu::close);
+    StorageException se2 =
+        assertThrows(StorageException.class, () -> ApiFutureUtils.await(finalObject));
+
+    assertAll(
+        () -> assertThat(se1).hasMessageThat().isEqualTo("Error: PERMISSION_DENIED"),
+        () -> assertThat(se2).hasMessageThat().isEqualTo("Error: PERMISSION_DENIED"),
+        () -> assertThat(se1.getCode()).isEqualTo(403),
+        () -> assertThat(se2.getCode()).isEqualTo(403));
+  }
+
+  @Test
+  public void badServerCrc32cResultsInException() throws Exception {
+    //noinspection resource
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.never(),
+            3,
+            finalObject,
+            new FakeStorageInternal() {
+              @Override
+              public BlobInfo compose(ComposeRequest composeRequest) {
+                BlobInfo response = super.compose(composeRequest);
+                // return a bad crc32c
+                return response.toBuilder().setCrc32c(Utils.crc32cCodec.encode(0)).build();
+              }
+            },
+            info,
+            opts);
+    pcu.write(DataGenerator.base64Characters().genByteBuffer(3));
+
+    AsynchronousCloseException se1 = assertThrows(AsynchronousCloseException.class, pcu::close);
+    StorageException se2 =
+        assertThrows(StorageException.class, () -> ApiFutureUtils.await(finalObject));
+
+    assertAll(
+        () -> assertThat(se1).hasCauseThat().isInstanceOf(StorageException.class),
+        () -> assertThat(se1).hasCauseThat().hasMessageThat().contains("Checksum mismatch"),
+        () -> assertThat(se2).hasMessageThat().contains("Checksum mismatch"),
+        () -> assertThat(se2.getCode()).isEqualTo(400));
+  }
+
+  @Test
+  public void bufferHandleRelease_returnsBufferOnFailureAndSuccess() throws Exception {
+    AtomicReference<Throwable> failure = new AtomicReference<>(null);
+    AtomicReference<String> success = new AtomicReference<>(null);
+
+    ApiFutureCallback<String> delegate =
+        new ApiFutureCallback<String>() {
+          @Override
+          public void onFailure(Throwable t) {
+            failure.set(t);
+          }
+
+          @Override
+          public void onSuccess(String result) {
+            success.set(result);
+          }
+        };
+
+    PooledBuffer p1 = PooledBuffer.of(BufferHandle.allocate(3));
+    BufferHandlePool pool =
+        new BufferHandlePool() {
+          @Override
+          public PooledBuffer getBuffer() {
+            return null;
+          }
+
+          @Override
+          public void returnBuffer(PooledBuffer handle) {
+            assertThat(handle).isSameInstanceAs(p1);
+          }
+        };
+
+    BufferHandleReleaser<String> releaser = new BufferHandleReleaser<>(pool, p1, delegate);
+
+    releaser.onSuccess("success");
+    releaser.onFailure(new Exception("induced failure"));
+
+    assertAll(
+        () -> assertThat(success.get()).isEqualTo("success"),
+        () -> assertThat(failure.get()).hasMessageThat().isEqualTo("induced failure"));
+  }
+
+  @Test
+  public void shortCircuitExceptionResultsInFastFailure() throws Exception {
+    ThreadFactory threadFactory =
+        new ThreadFactoryBuilder().setDaemon(true).setNameFormat("test-shortCircuit-%d").build();
+    ExecutorService threadPool = Executors.newFixedThreadPool(1, threadFactory);
+
+    try {
+
+      AtomicBoolean induceFailure = new AtomicBoolean(true);
+      CountDownLatch blockForWrite1 = new CountDownLatch(1);
+      CountDownLatch blockForWrite1Complete = new CountDownLatch(1);
+      FakeStorageInternal storageInternal =
+          new FakeStorageInternal() {
+            @Override
+            public BlobInfo internalDirectUpload(
+                BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+              if (induceFailure.getAndSet(false)) {
+                Uninterruptibles.awaitUninterruptibly(blockForWrite1);
+                try {
+                  throw StorageException.coalesce(
+                      ApiExceptionFactory.createException(
+                          "induced failure: " + info.getBlobId().toGsUtilUri(),
+                          null,
+                          GrpcStatusCode.of(Code.DATA_LOSS),
+                          false));
+                } finally {
+                  blockForWrite1Complete.countDown();
+                }
+              } else {
+                return super.internalDirectUpload(info, opts, buf);
+              }
+            }
+          };
+      ParallelCompositeUploadWritableByteChannel pcu =
+          new ParallelCompositeUploadWritableByteChannel(
+              bufferHandlePool,
+              threadPool,
+              partNamingStrategy,
+              PartCleanupStrategy.never(),
+              32,
+              finalObject,
+              storageInternal,
+              info,
+              opts);
+
+      byte[] bytes = DataGenerator.base64Characters().genBytes(bufferCapacity * 2 + 3);
+      // write the first parts worth of bytes
+      pcu.write(ByteBuffer.wrap(bytes, 0, bufferCapacity));
+      // signal that the blocking on the internalDirectUpload can proceed
+      blockForWrite1.countDown();
+      // wait until the internalDirectUpload has failed
+      blockForWrite1Complete.await();
+      // attempt to write some more bytes, where we should get a failure
+      StorageException storageException =
+          assertThrows(
+              StorageException.class,
+              () -> {
+                // due to the multiple threads doing uploads, it can sometimes take more than one
+                // invocation in order for the short circuit to trigger
+                for (int i = 0; i < 300; i++) {
+                  pcu.write(ByteBuffer.wrap(bytes, bufferCapacity, bufferCapacity));
+                }
+              });
+      // signal that the blocking on the internalDirectUpload can proceed
+      // blockForWrite2.countDown();
+      // wait until the internalDirectUpload has failed
+      // blockForWrite2Complete.await();
+      // calling close shouldn't cause another exception
+      pcu.close();
+
+      String name = info.getName();
+      BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), -1L);
+
+      CancellationException cancellationException =
+          assertThrows(
+              CancellationException.class,
+              () -> ApiExceptions.callAndTranslateApiException(finalObject));
+
+      assertAll(
+          () -> {
+            Optional<String> found =
+                storageInternal.addedObjects.keySet().stream()
+                    .map(BlobId::toGsUtilUri)
+                    .filter(p1.toGsUtilUri()::equals)
+                    .findFirst();
+            assertThat(found.isPresent()).isFalse();
+          },
+          () -> {
+            Optional<String> found =
+                storageInternal.addedObjects.keySet().stream()
+                    .map(BlobId::getName)
+                    .filter(name::equals)
+                    .findFirst();
+            assertThat(found.isPresent()).isFalse();
+          },
+          () ->
+              assertThat(storageException)
+                  .hasMessageThat()
+                  .contains("induced failure: " + p1.toGsUtilUri()),
+          () ->
+              assertThat(cancellationException)
+                  .hasCauseThat()
+                  .hasMessageThat()
+                  .contains("induced failure: " + p1.toGsUtilUri()));
+    } finally {
+      threadPool.shutdownNow();
+    }
+  }
+
+  @Test
+  public void errorContextIsPopulated() throws Exception {
+    //noinspection resource
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.never(),
+            3,
+            finalObject,
+            new FakeStorageInternal() {
+              @Override
+              public BlobInfo compose(ComposeRequest composeRequest) {
+                BlobInfo response = super.compose(composeRequest);
+                // return a bad crc32c
+                return response.toBuilder().setCrc32c(Utils.crc32cCodec.encode(0)).build();
+              }
+            },
+            info,
+            opts);
+    pcu.write(DataGenerator.base64Characters().genByteBuffer(3));
+
+    AsynchronousCloseException se1 = assertThrows(AsynchronousCloseException.class, pcu::close);
+    StorageException se2 =
+        assertThrows(StorageException.class, () -> ApiFutureUtils.await(finalObject));
+
+    String name = info.getName();
+    // parts 1-4
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    // ultimate object
+    BlobId expectedId = id(name, 2L);
+
+    assertAll(
+        () -> assertThat(se1).hasCauseThat().isInstanceOf(StorageException.class),
+        () -> assertThat(se1).hasCauseThat().hasMessageThat().contains("Checksum mismatch"),
+        () -> assertThat(se2).hasMessageThat().contains("Checksum mismatch"),
+        () -> {
+          assertThat(se2).hasCauseThat().isInstanceOf(ParallelCompositeUploadException.class);
+          ParallelCompositeUploadException pcue = (ParallelCompositeUploadException) se2.getCause();
+          // since we fail client side with a checksum validation, we expect the object to have been
+          // created
+          assertThat(pcue.getCreatedObjects().get()).containsExactly(p1, expectedId);
+        },
+        () -> assertThat(se2.getCode()).isEqualTo(400));
+  }
+
+  @Test
+  public void partFailedPreconditionOnRetryIsHandledGracefully() throws Exception {
+    String name = info.getName();
+    // parts 1-4
+    BlobId p1 = id(partNamingStrategy.fmtName(name, PartRange.of(1)), 1L);
+    BlobId p2 = id(partNamingStrategy.fmtName(name, PartRange.of(2)), 2L);
+    BlobId p3 = id(partNamingStrategy.fmtName(name, PartRange.of(3)), 3L);
+    // ultimate object
+    BlobId expectedId = id(name, 4L);
+
+    FakeStorageInternal storageInternal =
+        new FakeStorageInternal() {
+          @Override
+          public BlobInfo internalDirectUpload(
+              BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+            BlobInfo blobInfo = super.internalDirectUpload(info, opts, buf);
+            if (info.getName().equals(p1.getName())) {
+              throw StorageException.coalesce(
+                  ApiExceptionFactory.createException(
+                      null, GrpcStatusCode.of(Code.FAILED_PRECONDITION), false));
+            } else {
+              return blobInfo;
+            }
+          }
+
+          @Override
+          public BlobInfo internalObjectGet(BlobId blobId, Opts<ObjectSourceOpt> opts) {
+            Optional<BlobId> found = this.objectGet(blobId);
+            if (found.isPresent()) {
+              BlobId foundId = found.get();
+              Data d = this.addedObjects.get(foundId);
+              return d.getInfo();
+            }
+            throw StorageException.coalesce(
+                ApiExceptionFactory.createException(
+                    null, GrpcStatusCode.of(Code.NOT_FOUND), false));
+          }
+        };
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.always(),
+            10,
+            finalObject,
+            storageInternal,
+            info,
+            opts);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(bufferCapacity * 3 - 1);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+
+    BlobInfo result = ApiFutureUtils.await(finalObject);
+
+    assertAll(
+        () -> assertThat(result.getBlobId()).isEqualTo(expectedId),
+        () ->
+            assertThat(storageInternal.addedObjects.keySet())
+                .containsExactly(p1, p2, p3, expectedId),
+        () -> assertThat(storageInternal.deleteRequests).containsExactly(p1, p2, p3));
+  }
+
+  @NonNull
+  private ParallelCompositeUploadWritableByteChannel defaultPcu(int maxElementsPerCompact) {
+    return new ParallelCompositeUploadWritableByteChannel(
+        bufferHandlePool,
+        MoreExecutors.directExecutor(),
+        partNamingStrategy,
+        PartCleanupStrategy.always(),
+        maxElementsPerCompact,
+        finalObject,
+        storageInternal,
+        info,
+        opts);
+  }
+
+  private BlobId id(String name, long generation) {
+    return BlobId.of(info.getBucket(), name, generation);
+  }
+
+  private static class FakeStorageInternal implements StorageInternal {
+    protected final AtomicInteger generations;
+    protected final Map<BlobId, Data> addedObjects;
+    protected final List<ComposeRequest> composeRequests;
+    protected final List<BlobId> deleteRequests;
+
+    private FakeStorageInternal() {
+      generations = new AtomicInteger(1);
+      addedObjects = Collections.synchronizedMap(new HashMap<>());
+      composeRequests = Collections.synchronizedList(new ArrayList<>());
+      deleteRequests = Collections.synchronizedList(new ArrayList<>());
+    }
+
+    @Override
+    public BlobInfo internalDirectUpload(
+        BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+      BlobId id = info.getBlobId();
+
+      BlobInfo.Builder b = info.toBuilder();
+      WriteObjectRequest apply =
+          opts.writeObjectRequest().apply(WriteObjectRequest.newBuilder()).build();
+      if (apply.hasWriteObjectSpec() && apply.getWriteObjectSpec().hasIfGenerationMatch()) {
+        long ifGenerationMatch = apply.getWriteObjectSpec().getIfGenerationMatch();
+        Optional<BlobId> existing = objectGet(id);
+        if (existing.isPresent()) {
+          BlobId existingId = existing.get();
+          if (ifGenerationMatch != existingId.getGeneration()) {
+            throw StorageException.coalesce(
+                ApiExceptionFactory.createException(
+                    null, GrpcStatusCode.of(Code.FAILED_PRECONDITION), false));
+          }
+        }
+      }
+      BlobId newId = id.withGeneration(generations.getAndIncrement());
+      b.setBlobId(newId);
+      BlobInfo gen1 = b.build();
+      addedObjects.put(newId, new Data(gen1, HASHER.hash(buf)));
+      return gen1;
+    }
+
+    @NonNull
+    protected Optional<BlobId> objectGet(BlobId id) {
+      return addedObjects.keySet().stream()
+          .filter(
+              key -> key.getBucket().equals(id.getBucket()) && key.getName().equals(id.getName()))
+          .findFirst();
+    }
+
+    @Override
+    public BlobInfo compose(ComposeRequest composeRequest) {
+      composeRequests.add(composeRequest);
+      BlobInfo info = composeRequest.getTarget();
+      String bucket = info.getBucket();
+      BlobInfo.Builder b = info.toBuilder();
+      BlobId newId = info.getBlobId().withGeneration(generations.getAndIncrement());
+      b.setBlobId(newId);
+      ImmutableList<Crc32cLengthKnown> crc32cs =
+          composeRequest.getSourceBlobs().stream()
+              .map(so -> BlobId.of(bucket, so.getName(), so.getGeneration()))
+              .map(addedObjects::get)
+              .map(Data::getCrc32c)
+              .collect(ImmutableList.toImmutableList());
+
+      Crc32cLengthKnown reduce = crc32cs.stream().reduce(null, HASHER::nullSafeConcat);
+      Preconditions.checkState(reduce != null, "unable to compute crc32c for compose request");
+      b.setCrc32c(Utils.crc32cCodec.encode(reduce.getValue()));
+      BlobInfo gen1 = b.build();
+      addedObjects.put(newId, new Data(gen1, reduce));
+      return gen1;
+    }
+
+    @Override
+    public boolean delete(BlobId id) {
+      deleteRequests.add(id);
+      return addedObjects.containsKey(id);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("generations", generations)
+          .add("addedObjects", addedObjects)
+          .add("composeRequests", composeRequests)
+          .add("deleteRequests", deleteRequests)
+          .toString();
+    }
+
+    protected static final class Data {
+      private final BlobInfo info;
+      private final Crc32cLengthKnown crc32c;
+
+      private Data(BlobInfo info, Crc32cLengthKnown crc32c) {
+        this.info = info;
+        this.crc32c = crc32c;
+      }
+
+      public BlobInfo getInfo() {
+        return info;
+      }
+
+      public Crc32cLengthKnown getCrc32c() {
+        return crc32c;
+      }
+    }
+  }
+
+  private static class SimplisticPartNamingStrategy extends PartNamingStrategy {
+
+    private final String prefix;
+
+    private SimplisticPartNamingStrategy(String prefix) {
+      super(null);
+      this.prefix = prefix;
+    }
+
+    @Override
+    String fmtName(String ultimateObjectName, PartRange partRange) {
+      return String.format("%s/%s/%s.part", prefix, ultimateObjectName, partRange.encode());
+    }
+
+    @Override
+    protected String fmtFields(String randomKey, String nameDigest, String partRange) {
+      return null;
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java
index 400d7e19..68efe4f8 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java
@@ -20,15 +20,19 @@ 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.UnifiedOpts.ObjectSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import javax.crypto.SecretKey;
 import org.junit.Test;
 import org.junit.runners.model.MultipleFailureException;
 
@@ -141,6 +145,41 @@ public final class UnifiedOptsTest {
     validateFactoryMethodEnforceNonNull(Storage.UpdateHmacKeyOption.class);
   }
 
+  @Test
+  public void transformTo() {
+    SecretKey key =
+        new SecretKey() {
+          @Override
+          public String getAlgorithm() {
+            return "fake";
+          }
+
+          @Override
+          public String getFormat() {
+            return null;
+          }
+
+          @Override
+          public byte[] getEncoded() {
+            return "fake".getBytes(StandardCharsets.UTF_8);
+          }
+        };
+
+    Opts<ObjectTargetOpt> targetOpts =
+        Opts.from(
+            // encryptionKey should project as a decryptionKey
+            UnifiedOpts.encryptionKey(key),
+            // userProject implements both target and source
+            UnifiedOpts.userProject("user-project"),
+            // crc32c is not a source opt or a ProjectToSource opt, it should be excluded
+            UnifiedOpts.crc32cMatch(1));
+    Opts<ObjectSourceOpt> sourceOpts = targetOpts.transformTo(ObjectSourceOpt.class);
+
+    Opts<ObjectSourceOpt> expected =
+        Opts.from(UnifiedOpts.decryptionKey(key), UnifiedOpts.userProject("user-project"));
+    assertThat(sourceOpts).isEqualTo(expected);
+  }
+
   private static void validateFactoryMethodEnforceNonNull(Class<?> classToSearch) throws Exception {
     validateFactoryMethodEnforceNonNull(classToSearch, classToSearch);
   }