feat: implement deleteSourceObjects for compose operation

Added deleteSourceObjects to Storage.ComposeRequest and Builder.
Updated StorageImpl, GrpcStorageImpl and HttpStorageRpc to support the flag.
Added unit test in StorageImplMockitoTest and integration test in ITObjectTest.

Co-authored-by: nidhiii-27 <224584462+nidhiii-27@users.noreply.github.com>

Author: google-labs-jules[bot]

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java

@@ -645,6 +645,9 @@ public Blob compose(ComposeRequest composeRequest) {
         .forEach(builder::addSourceObjects);
     final Object target = codecs.blobInfo().encode(composeRequest.getTarget());
     builder.setDestination(target);
+    if (composeRequest.isDeleteSourceObjects()) {
+      builder.setDeleteSourceObjects(true);
+    }
     ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return retrier.run(

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java

@@ -3185,6 +3185,7 @@ class ComposeRequest implements Serializable {
     private final List<SourceBlob> sourceBlobs;
     private final BlobInfo target;
     private final List<BlobTargetOption> targetOptions;
+    private final boolean deleteSourceObjects;
 
     private transient Opts<ObjectTargetOpt> targetOpts;
 
@@ -3222,6 +3223,7 @@ public static class Builder {
       private final Set<BlobTargetOption> targetOptions = new LinkedHashSet<>();
       private BlobInfo target;
       private Opts<ObjectTargetOpt> opts = Opts.empty();
+      private boolean deleteSourceObjects;
 
       /** Add source blobs for compose operation. */
       public Builder addSource(Iterable<String> blobs) {
@@ -3265,6 +3267,16 @@ public Builder setTargetOptions(Iterable<BlobTargetOption> options) {
         return this;
       }
 
+      /**
+       * Sets whether to delete source blobs after compose operation.
+       *
+       * @since 2.67.0
+       */
+      public Builder setDeleteSourceObjects(boolean deleteSourceObjects) {
+        this.deleteSourceObjects = deleteSourceObjects;
+        return this;
+      }
+
       /** Creates a {@code ComposeRequest} object. */
       public ComposeRequest build() {
         checkArgument(!sourceBlobs.isEmpty());
@@ -3280,6 +3292,7 @@ private ComposeRequest(Builder builder) {
       // keep targetOptions for serialization even though we will read targetOpts
       targetOptions = ImmutableList.copyOf(builder.targetOptions);
       targetOpts = builder.opts.prepend(Opts.unwrap(targetOptions).resolveFrom(target));
+      deleteSourceObjects = builder.deleteSourceObjects;
     }
 
     /** Returns compose operation's source blobs. */
@@ -3297,6 +3310,11 @@ public List<BlobTargetOption> getTargetOptions() {
       return targetOptions;
     }
 
+    /** Returns whether to delete source blobs after compose operation. */
+    public boolean isDeleteSourceObjects() {
+      return deleteSourceObjects;
+    }
+
     @InternalApi
     Opts<ObjectTargetOpt> getTargetOpts() {
       return targetOpts;

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java

@@ -651,7 +651,10 @@ public Blob compose(final ComposeRequest composeRequest) {
     }
     Opts<ObjectTargetOpt> targetOpts = composeRequest.getTargetOpts();
     StorageObject targetPb = codecs.blobInfo().encode(composeRequest.getTarget());
-    Map<StorageRpc.Option, ?> targetOptions = targetOpts.getRpcOptions();
+    Map<StorageRpc.Option, Object> targetOptions = Maps.newHashMap(targetOpts.getRpcOptions());
+    if (composeRequest.isDeleteSourceObjects()) {
+      targetOptions.put(StorageRpc.Option.DELETE_SOURCE_OBJECTS, true);
+    }
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions);
     return run(

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java

@@ -817,6 +817,9 @@ public StorageObject compose(
       sourceObjects.add(sourceObject);
     }
     request.setSourceObjects(sourceObjects);
+    if (Option.DELETE_SOURCE_OBJECTS.getBoolean(targetOptions) != null) {
+      request.setDeleteSourceObjects(Option.DELETE_SOURCE_OBJECTS.getBoolean(targetOptions));
+    }
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_COMPOSE);
     Scope scope = tracer.withSpan(span);
     try {

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java

@@ -82,6 +82,7 @@ enum Option {
     INCLUDE_TRAILING_DELIMITER("includeTrailingDelimiter"),
     X_UPLOAD_CONTENT_LENGTH("x-upload-content-length"),
     OBJECT_FILTER("objectFilter"),
+    DELETE_SOURCE_OBJECTS("deleteSourceObjects"),
     /**
      * An {@link com.google.common.collect.ImmutableMap ImmutableMap<String, String>} of values
      * which will be set as additional headers on the request.

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java

@@ -1038,4 +1038,39 @@ private void verifyBucketNotification(Notification value) {
     assertEquals(TOPIC, value.getTopic());
     assertEquals(Arrays.asList(EVENT_TYPES), value.getEventTypes());
   }
+
+  @Test
+  public void testComposeWithDeleteSourceObjects() {
+    String bucket = "b1";
+    String source1 = "s1";
+    String source2 = "s2";
+    String target = "t1";
+    BlobId targetId = BlobId.of(bucket, target);
+    BlobInfo targetInfo = BlobInfo.newBuilder(targetId).build();
+    Storage.ComposeRequest req =
+        Storage.ComposeRequest.newBuilder()
+            .addSource(source1, source2)
+            .setTarget(targetInfo)
+            .setDeleteSourceObjects(true)
+            .build();
+
+    StorageObject targetPb = Conversions.json().blobInfo().encode(targetInfo);
+    List<StorageObject> sourcePbs =
+        ImmutableList.of(
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(bucket, source1).build()),
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(bucket, source2).build()));
+
+    ArgumentCaptor<Map<StorageRpc.Option, Object>> optionsCaptor =
+        ArgumentCaptor.forClass(Map.class);
+
+    doReturn(targetPb)
+        .when(storageRpcMock)
+        .compose(Mockito.eq(sourcePbs), Mockito.eq(targetPb), optionsCaptor.capture());
+
+    initializeService();
+    storage.compose(req);
+
+    Map<StorageRpc.Option, Object> capturedOptions = optionsCaptor.getValue();
+    assertEquals(true, capturedOptions.get(StorageRpc.Option.DELETE_SOURCE_OBJECTS));
+  }
 }

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java

@@ -812,6 +812,34 @@ public void testComposeBlobWithContentType() {
     assertArrayEquals(composedBytes, readBytes);
   }
 
+  @Test
+  public void testComposeBlobWithDeleteSourceObjects() {
+    String baseName = generator.randomObjectName();
+    String sourceBlobName1 = baseName + "-1";
+    String sourceBlobName2 = baseName + "-2";
+    BlobInfo sourceBlob1 = BlobInfo.newBuilder(bucket, sourceBlobName1).build();
+    BlobInfo sourceBlob2 = BlobInfo.newBuilder(bucket, sourceBlobName2).build();
+    storage.create(sourceBlob1, BLOB_BYTE_CONTENT);
+    storage.create(sourceBlob2, BLOB_BYTE_CONTENT);
+
+    String targetBlobName = baseName + "-target";
+    BlobInfo targetBlob = BlobInfo.newBuilder(bucket, targetBlobName).build();
+    ComposeRequest req =
+        ComposeRequest.newBuilder()
+            .addSource(sourceBlobName1, sourceBlobName2)
+            .setTarget(targetBlob)
+            .setDeleteSourceObjects(true)
+            .build();
+    Blob remoteTargetBlob = storage.compose(req);
+    assertNotNull(remoteTargetBlob);
+
+    assertNull(storage.get(bucket.getName(), sourceBlobName1));
+    assertNull(storage.get(bucket.getName(), sourceBlobName2));
+
+    byte[] readBytes = storage.readAllBytes(bucket.getName(), targetBlobName);
+    assertThat(readBytes.length).isEqualTo(BLOB_BYTE_CONTENT.length * 2);
+  }
+
   @Test
   public void testComposeBlobFail() {
     String baseName = generator.randomObjectName();