Repository: googleapis/java-storage
Pull request #2434 (2024-03-04 19:59:38)

feat: Add Custom Part Metadata Decorator to ParallelCompositeUploadConfig


Patch:
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 e13f5496..07551abe 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
@@ -42,12 +42,15 @@ import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
 import java.security.SecureRandom;
 import java.time.Clock;
+import java.time.Duration;
+import java.time.OffsetDateTime;
 import java.util.Base64;
 import java.util.Base64.Encoder;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.UnaryOperator;
 import javax.annotation.concurrent.Immutable;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
@@ -125,18 +128,21 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
   private final BufferAllocationStrategy bufferAllocationStrategy;
   private final PartNamingStrategy partNamingStrategy;
   private final PartCleanupStrategy partCleanupStrategy;
+  private final PartMetadataFieldDecorator partMetadataFieldDecorator;
 
   private ParallelCompositeUploadBlobWriteSessionConfig(
       int maxPartsPerCompose,
       ExecutorSupplier executorSupplier,
       BufferAllocationStrategy bufferAllocationStrategy,
       PartNamingStrategy partNamingStrategy,
-      PartCleanupStrategy partCleanupStrategy) {
+      PartCleanupStrategy partCleanupStrategy,
+      PartMetadataFieldDecorator partMetadataFieldDecorator) {
     this.maxPartsPerCompose = maxPartsPerCompose;
     this.executorSupplier = executorSupplier;
     this.bufferAllocationStrategy = bufferAllocationStrategy;
     this.partNamingStrategy = partNamingStrategy;
     this.partCleanupStrategy = partCleanupStrategy;
+    this.partMetadataFieldDecorator = partMetadataFieldDecorator;
   }
 
   @InternalApi
@@ -150,7 +156,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         executorSupplier,
         bufferAllocationStrategy,
         partNamingStrategy,
-        partCleanupStrategy);
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
   }
 
   /**
@@ -170,7 +177,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         executorSupplier,
         bufferAllocationStrategy,
         partNamingStrategy,
-        partCleanupStrategy);
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
   }
 
   /**
@@ -191,7 +199,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         executorSupplier,
         bufferAllocationStrategy,
         partNamingStrategy,
-        partCleanupStrategy);
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
   }
 
   /**
@@ -211,7 +220,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         executorSupplier,
         bufferAllocationStrategy,
         partNamingStrategy,
-        partCleanupStrategy);
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
   }
 
   /**
@@ -231,7 +241,29 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         executorSupplier,
         bufferAllocationStrategy,
         partNamingStrategy,
-        partCleanupStrategy);
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
+  }
+
+  /**
+   * Specify a Part Metadata Field decorator, this will manipulate the metadata associated with part
+   * objects, the ultimate object metadata will remain unchanged.
+   *
+   * <p><i>Default: </i> {@link PartMetadataFieldDecorator#noOp()}
+   *
+   * @since 2.36.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  public ParallelCompositeUploadBlobWriteSessionConfig withPartMetadataFieldDecorator(
+      PartMetadataFieldDecorator partMetadataFieldDecorator) {
+    checkNotNull(partMetadataFieldDecorator, "partMetadataFieldDecorator must be non null");
+    return new ParallelCompositeUploadBlobWriteSessionConfig(
+        maxPartsPerCompose,
+        executorSupplier,
+        bufferAllocationStrategy,
+        partNamingStrategy,
+        partCleanupStrategy,
+        partMetadataFieldDecorator);
   }
 
   @BetaApi
@@ -241,7 +273,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
         ExecutorSupplier.cachedPool(),
         BufferAllocationStrategy.simple(ByteSizeConstants._16MiB),
         PartNamingStrategy.noPrefix(),
-        PartCleanupStrategy.always());
+        PartCleanupStrategy.always(),
+        PartMetadataFieldDecorator.noOp());
   }
 
   @InternalApi
@@ -249,7 +282,10 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
   WriterFactory createFactory(Clock clock) throws IOException {
     Executor executor = executorSupplier.get();
     BufferHandlePool bufferHandlePool = bufferAllocationStrategy.get();
-    return new ParallelCompositeUploadWriterFactory(clock, executor, bufferHandlePool);
+    PartMetadataFieldDecoratorInstance partMetadataFieldDecoratorInstance =
+        partMetadataFieldDecorator.newInstance(clock);
+    return new ParallelCompositeUploadWriterFactory(
+        clock, executor, bufferHandlePool, partMetadataFieldDecoratorInstance);
   }
 
   /**
@@ -277,6 +313,7 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
      */
     @BetaApi
     public static BufferAllocationStrategy simple(int capacity) {
+      checkArgument(capacity > 0, "bufferCapacity must be > 0");
       return new SimpleBufferAllocationStrategy(capacity);
     }
 
@@ -291,6 +328,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
      */
     @BetaApi
     public static BufferAllocationStrategy fixedPool(int bufferCount, int bufferCapacity) {
+      checkArgument(bufferCount > 0, "bufferCount must be > 0");
+      checkArgument(bufferCapacity > 0, "bufferCapacity must be > 0");
       return new FixedPoolBufferAllocationStrategy(bufferCount, bufferCapacity);
     }
 
@@ -361,6 +400,7 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
      */
     @BetaApi
     public static ExecutorSupplier fixedPool(int poolSize) {
+      checkArgument(poolSize > 0, "poolSize must be > 0");
       return new FixedSupplier(poolSize);
     }
 
@@ -631,6 +671,79 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
     }
   }
 
+  /**
+   * A Decorator which is used to manipulate metadata fields, specifically on the part objects
+   * created in a Parallel Composite Upload
+   *
+   * @see #withPartMetadataFieldDecorator(PartMetadataFieldDecorator)
+   * @since 2.36.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  @Immutable
+  public abstract static class PartMetadataFieldDecorator implements Serializable {
+
+    abstract PartMetadataFieldDecoratorInstance newInstance(Clock clock);
+
+    /**
+     * A decorator that is used to manipulate the Custom Time Metadata field of part files. {@link
+     * BlobInfo#getCustomTimeOffsetDateTime()}
+     *
+     * <p>When provided with a duration, a time in the future will be calculated for each part
+     * object upon upload, this new value can be used in OLM rules to cleanup abandoned part files.
+     *
+     * <p>See [CustomTime OLM
+     * documentation](https://cloud.google.com/storage/docs/lifecycle#dayssincecustomtime)
+     *
+     * @see #withPartMetadataFieldDecorator(PartMetadataFieldDecorator)
+     * @since 2.36.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public static PartMetadataFieldDecorator setCustomTimeInFuture(Duration timeInFuture) {
+      checkNotNull(timeInFuture, "timeInFuture must not be null");
+      return new CustomTimeInFuture(timeInFuture);
+    }
+
+    @BetaApi
+    public static PartMetadataFieldDecorator noOp() {
+      return NoOp.INSTANCE;
+    }
+
+    @BetaApi
+    private static final class CustomTimeInFuture extends PartMetadataFieldDecorator {
+      private static final long serialVersionUID = -6213742554954751892L;
+      private final Duration duration;
+
+      CustomTimeInFuture(Duration duration) {
+        this.duration = duration;
+      }
+
+      @Override
+      PartMetadataFieldDecoratorInstance newInstance(Clock clock) {
+        return builder -> {
+          OffsetDateTime futureTime =
+              OffsetDateTime.from(
+                  clock.instant().plus(duration).atZone(clock.getZone()).toOffsetDateTime());
+          return builder.setCustomTimeOffsetDateTime(futureTime);
+        };
+      }
+    }
+
+    private static final class NoOp extends PartMetadataFieldDecorator {
+      private static final long serialVersionUID = -4569486383992999205L;
+      private static final NoOp INSTANCE = new NoOp();
+
+      @Override
+      PartMetadataFieldDecoratorInstance newInstance(Clock clock) {
+        return builder -> builder;
+      }
+
+      /** prevent java serialization from using a new instance */
+      private Object readResolve() {
+        return INSTANCE;
+      }
+    }
+  }
+
   /**
    * A cleanup strategy which will dictate what cleanup operations are performed automatically when
    * performing a parallel composite upload.
@@ -708,6 +821,8 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
     }
   }
 
+  interface PartMetadataFieldDecoratorInstance extends UnaryOperator<BlobInfo.Builder> {}
+
   private abstract static class Factory<T> implements Serializable {
     private static final long serialVersionUID = 271806144227661056L;
 
@@ -721,12 +836,17 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
     private final Clock clock;
     private final Executor executor;
     private final BufferHandlePool bufferHandlePool;
+    private final PartMetadataFieldDecoratorInstance partMetadataFieldDecoratorInstance;
 
     private ParallelCompositeUploadWriterFactory(
-        Clock clock, Executor executor, BufferHandlePool bufferHandlePool) {
+        Clock clock,
+        Executor executor,
+        BufferHandlePool bufferHandlePool,
+        PartMetadataFieldDecoratorInstance partMetadataFieldDecoratorInstance) {
       this.clock = clock;
       this.executor = executor;
       this.bufferHandlePool = bufferHandlePool;
+      this.partMetadataFieldDecoratorInstance = partMetadataFieldDecoratorInstance;
     }
 
     @Override
@@ -760,6 +880,7 @@ public final class ParallelCompositeUploadBlobWriteSessionConfig extends BlobWri
                 partNamingStrategy,
                 partCleanupStrategy,
                 maxPartsPerCompose,
+                partMetadataFieldDecoratorInstance,
                 result,
                 storageInternal,
                 info,
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
index 9ff1ebdb..639802cd 100644
--- 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
@@ -31,6 +31,7 @@ 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.PartMetadataFieldDecoratorInstance;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
 import com.google.cloud.storage.Storage.ComposeRequest;
 import com.google.cloud.storage.UnifiedOpts.Crc32cMatch;
@@ -111,6 +112,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
   private final PartNamingStrategy partNamingStrategy;
   private final PartCleanupStrategy partCleanupStrategy;
   private final int maxElementsPerCompact;
+  private final PartMetadataFieldDecoratorInstance partMetadataFieldDecorator;
   private final SettableApiFuture<BlobInfo> finalObject;
   private final StorageInternal storage;
   private final BlobInfo ultimateObject;
@@ -135,6 +137,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
       PartNamingStrategy partNamingStrategy,
       PartCleanupStrategy partCleanupStrategy,
       int maxElementsPerCompact,
+      PartMetadataFieldDecoratorInstance partMetadataFieldDecorator,
       SettableApiFuture<BlobInfo> finalObject,
       StorageInternal storage,
       BlobInfo ultimateObject,
@@ -144,6 +147,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
     this.partNamingStrategy = partNamingStrategy;
     this.partCleanupStrategy = partCleanupStrategy;
     this.maxElementsPerCompact = maxElementsPerCompact;
+    this.partMetadataFieldDecorator = partMetadataFieldDecorator;
     this.finalObject = finalObject;
     this.storage = storage;
     this.ultimateObject = ultimateObject;
@@ -427,6 +431,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
     PART_INDEX.appendTo(partRange, builder);
     OBJECT_OFFSET.appendTo(offset, builder);
     b.setMetadata(builder.build());
+    b = partMetadataFieldDecorator.apply(b);
     return b.build();
   }
 


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfigTest.java
index 465741a7..3ef55372 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfigTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ParallelCompositeUploadBlobWriteSessionConfigTest.java
@@ -21,8 +21,13 @@ import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 
 import com.google.cloud.storage.MetadataField.PartRange;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartMetadataFieldDecorator;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
 import com.google.common.truth.StringSubject;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
 import org.junit.Test;
 
 public final class ParallelCompositeUploadBlobWriteSessionConfigTest {
@@ -87,6 +92,18 @@ public final class ParallelCompositeUploadBlobWriteSessionConfigTest {
         () -> assertThat(fmt).startsWith("a/b/obj"));
   }
 
+  @Test
+  public void partMetadataFieldDecorator_customTime() {
+    BlobInfo.Builder testBlob = BlobInfo.newBuilder("testBlob", "testBucket");
+    Duration duration = Duration.ofSeconds(30);
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+    OffsetDateTime expected =
+        OffsetDateTime.from(Instant.EPOCH.plus(duration).atZone(ZoneId.of("Z")));
+    PartMetadataFieldDecorator.setCustomTimeInFuture(duration).newInstance(clock).apply(testBlob);
+
+    assertThat(expected).isEqualTo(testBlob.build().getCustomTimeOffsetDateTime());
+  }
+
   private static StringSubject assertField(String fmt, int idx) {
     String[] split = fmt.split(";");
     String s = split[idx];
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
index ed66a3f1..3a78adf4 100644
--- 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
@@ -29,6 +29,8 @@ 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.PartMetadataFieldDecorator;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartMetadataFieldDecoratorInstance;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
 import com.google.cloud.storage.ParallelCompositeUploadWritableByteChannel.BufferHandleReleaser;
 import com.google.cloud.storage.Storage.ComposeRequest;
@@ -47,9 +49,14 @@ 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.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.ClosedChannelException;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.Collections;
@@ -80,6 +87,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
   private SettableApiFuture<BlobInfo> finalObject;
   private FakeStorageInternal storageInternal;
   private SimplisticPartNamingStrategy partNamingStrategy;
+  private PartMetadataFieldDecoratorInstance partMetadataFieldDecorator;
   private int bufferCapacity;
 
   @Before
@@ -91,6 +99,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
     finalObject = SettableApiFuture.create();
     partNamingStrategy = new SimplisticPartNamingStrategy("prefix");
     storageInternal = new FakeStorageInternal();
+    partMetadataFieldDecorator = PartMetadataFieldDecorator.noOp().newInstance(null);
   }
 
   @Test
@@ -202,6 +211,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.never(),
             maxElementsPerCompact,
+            partMetadataFieldDecorator,
             finalObject,
             storageInternal,
             info,
@@ -241,6 +251,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.never(),
             maxElementsPerCompact,
+            partMetadataFieldDecorator,
             finalObject,
             storageInternal,
             info,
@@ -340,6 +351,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.never(),
             3,
+            partMetadataFieldDecorator,
             finalObject,
             new FakeStorageInternal() {
               @Override
@@ -429,6 +441,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.always(),
             3,
+            partMetadataFieldDecorator,
             finalObject,
             new FakeStorageInternal() {
               @Override
@@ -462,6 +475,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.always(),
             3,
+            partMetadataFieldDecorator,
             finalObject,
             new FakeStorageInternal() {
               @Override
@@ -568,6 +582,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
               partNamingStrategy,
               PartCleanupStrategy.never(),
               32,
+              partMetadataFieldDecorator,
               finalObject,
               storageInternal,
               info,
@@ -647,6 +662,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.never(),
             3,
+            partMetadataFieldDecorator,
             finalObject,
             new FakeStorageInternal() {
               @Override
@@ -729,6 +745,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
             partNamingStrategy,
             PartCleanupStrategy.always(),
             10,
+            partMetadataFieldDecorator,
             finalObject,
             storageInternal,
             info,
@@ -749,6 +766,49 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
         () -> assertThat(storageInternal.deleteRequests).containsExactly(p1, p2, p3));
   }
 
+  @Test
+  public void partMetadataFieldDecoratorUsesCustomTime() throws IOException {
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+    OffsetDateTime rangeBegin =
+        OffsetDateTime.from(Instant.EPOCH.plus(Duration.ofSeconds(29)).atZone(ZoneId.of("Z")));
+    OffsetDateTime rangeEnd =
+        OffsetDateTime.from(Instant.EPOCH.plus(Duration.ofMinutes(2)).atZone(ZoneId.of("Z")));
+
+    FakeStorageInternal storageInternal =
+        new FakeStorageInternal() {
+          @Override
+          public BlobInfo internalDirectUpload(
+              BlobInfo info, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
+            if (info.getBlobId().getName().endsWith(".part")) {
+              // Kinda hacky but since we are creating multiple parts we will use a range
+              // to ensure the customTimes are being calculated appropriately
+              assertThat(info.getCustomTimeOffsetDateTime().isAfter(rangeBegin)).isTrue();
+              assertThat(info.getCustomTimeOffsetDateTime().isBefore(rangeEnd)).isTrue();
+            } else {
+              assertThat(info.getCustomTimeOffsetDateTime()).isNull();
+            }
+            return super.internalDirectUpload(info, opts, buf);
+          }
+        };
+    ParallelCompositeUploadWritableByteChannel pcu =
+        new ParallelCompositeUploadWritableByteChannel(
+            bufferHandlePool,
+            MoreExecutors.directExecutor(),
+            partNamingStrategy,
+            PartCleanupStrategy.always(),
+            10,
+            PartMetadataFieldDecorator.setCustomTimeInFuture(Duration.ofSeconds(30))
+                .newInstance(clock),
+            finalObject,
+            storageInternal,
+            info,
+            opts);
+    byte[] bytes = DataGenerator.base64Characters().genBytes(bufferCapacity * 3 - 1);
+    pcu.write(ByteBuffer.wrap(bytes));
+
+    pcu.close();
+  }
+
   @NonNull
   private ParallelCompositeUploadWritableByteChannel defaultPcu(int maxElementsPerCompact) {
     return new ParallelCompositeUploadWritableByteChannel(
@@ -757,6 +817,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
         partNamingStrategy,
         PartCleanupStrategy.always(),
         maxElementsPerCompact,
+        partMetadataFieldDecorator,
         finalObject,
         storageInternal,
         info,
@@ -773,7 +834,7 @@ public final class ParallelCompositeUploadWritableByteChannelTest {
     protected final List<ComposeRequest> composeRequests;
     protected final List<BlobId> deleteRequests;
 
-    private FakeStorageInternal() {
+    FakeStorageInternal() {
       generations = new AtomicInteger(1);
       addedObjects = Collections.synchronizedMap(new HashMap<>());
       composeRequests = Collections.synchronizedList(new ArrayList<>());
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index eab0e78e..5f642b50 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -36,6 +36,7 @@ import com.google.cloud.storage.BlobWriteChannelV2.BlobWriteChannelV2State;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferAllocationStrategy;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartMetadataFieldDecorator;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BucketField;
@@ -55,6 +56,7 @@ import java.io.InvalidClassException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.Map;
@@ -388,10 +390,16 @@ public class SerializationTest extends BaseSerializationTest {
             .withBufferAllocationStrategy(BufferAllocationStrategy.fixedPool(1, 3))
             .withPartCleanupStrategy(PartCleanupStrategy.never())
             .withPartNamingStrategy(PartNamingStrategy.prefix("prefix"))
-            .withExecutorSupplier(ExecutorSupplier.fixedPool(5));
+            .withExecutorSupplier(ExecutorSupplier.fixedPool(5))
+            .withPartMetadataFieldDecorator(
+                PartMetadataFieldDecorator.setCustomTimeInFuture(Duration.ofMinutes(10)));
     ParallelCompositeUploadBlobWriteSessionConfig pcu2copy = serializeAndDeserialize(pcu2);
     assertThat(pcu2copy).isNotNull();
 
+    PartMetadataFieldDecorator noop = PartMetadataFieldDecorator.noOp();
+    PartMetadataFieldDecorator noopCopy = serializeAndDeserialize(noop);
+    assertThat(noopCopy).isSameInstanceAs(noop);
+
     InvalidClassException invalidClassException =
         assertThrows(
             InvalidClassException.class,
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java
index e37a4f02..89d04c35 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java
@@ -37,7 +37,7 @@ final class TestClock extends Clock {
 
   @Override
   public ZoneId getZone() {
-    throw new UnsupportedOperationException("TestClock.getZone()");
+    return ZoneId.of("Z");
   }
 
   @Override
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
index 012a700f..304cdd69 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITParallelCompositeUploadBlobWriteSessionConfigTest.java
@@ -21,6 +21,7 @@ import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
 
+import com.google.api.gax.paging.Page;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.cloud.kms.v1.CryptoKey;
 import com.google.cloud.storage.Blob;
@@ -34,6 +35,7 @@ import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.BufferAllocationStrategy;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.ExecutorSupplier;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartCleanupStrategy;
+import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartMetadataFieldDecorator;
 import com.google.cloud.storage.ParallelCompositeUploadBlobWriteSessionConfig.PartNamingStrategy;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobSourceOption;
@@ -57,6 +59,7 @@ import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.WritableByteChannel;
 import java.security.Key;
+import java.time.Duration;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -109,8 +112,11 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
             // define a max part size that is fairly small to aid in test speed
             .withBufferAllocationStrategy(BufferAllocationStrategy.simple(_1MiB))
             .withPartNamingStrategy(PartNamingStrategy.prefix("prefix-a"))
-            // let our fixtures take care of cleaning things up if an upload fails
-            .withPartCleanupStrategy(PartCleanupStrategy.onlyOnSuccess());
+            // Write customTime 30 seconds in the future
+            .withPartMetadataFieldDecorator(
+                PartMetadataFieldDecorator.setCustomTimeInFuture(Duration.ofSeconds(30)))
+            // let our fixtures take care of cleaning things
+            .withPartCleanupStrategy(PartCleanupStrategy.never());
 
     StorageOptions storageOptions = null;
     if (transport == Transport.GRPC) {
@@ -140,6 +146,15 @@ public final class ITParallelCompositeUploadBlobWriteSessionConfigTest {
     }
   }
 
+  @Test
+  public void partFilesCreatedWithCustomTimeWritten() throws IOException {
+    doTest(bucket, 10 * _1MiB + 37, ImmutableList.of(), ImmutableList.of(), ImmutableList.of());
+    Page<Blob> blobs = storage.list(bucket.getName(), Storage.BlobListOption.prefix("prefix-a"));
+    for (Blob blob : blobs.iterateAll()) {
+      assertThat(blob.getCustomTimeOffsetDateTime()).isNotNull();
+    }
+  }
+
   @Test
   public void errorRaisedByMethodAndFutureResult() throws IOException {