diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index b536340e9c23..120b7a269724 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -645,6 +645,7 @@ public Blob compose(ComposeRequest composeRequest) {
         .forEach(builder::addSourceObjects);
     final Object target = codecs.blobInfo().encode(composeRequest.getTarget());
     builder.setDestination(target);
+    builder.setDeleteSourceObjects(composeRequest.isDeleteSourceObjects());
     ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return retrier.run(
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 12ac95dff7c4..9d71d8f73711 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/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;
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index ebc4cbe5d736..8a510b84e7cc 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/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(
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index 5f910fb7775a..97814b597c37 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -817,6 +817,7 @@ public StorageObject compose(
       sourceObjects.add(sourceObject);
     }
     request.setSourceObjects(sourceObjects);
+    request.setDeleteSourceObjects(Option.DELETE_SOURCE_OBJECTS.getBoolean(targetOptions));
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_COMPOSE);
     Scope scope = tracer.withSpan(span);
     try {
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
index 59a56df12022..33f24a54c854 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
+++ b/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&lt;String, String>} of values
      * which will be set as additional headers on the request.
diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
index ec9dfe9e7460..5651b8e2de57 100644
--- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
+++ b/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));
+  }
 }
diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
index 83a7aeb5a912..f186afd077b6 100644
--- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
+++ b/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();