diff --git a/google/cloud/storage/internal/grpc/object_request_parser.cc b/google/cloud/storage/internal/grpc/object_request_parser.cc
index 5c6e1027a2d72..4ef0ab303f973 100644
--- a/google/cloud/storage/internal/grpc/object_request_parser.cc
+++ b/google/cloud/storage/internal/grpc/object_request_parser.cc
@@ -371,6 +371,9 @@ StatusOr<google::storage::v2::ComposeObjectRequest> ToProto(
         request.GetOption<storage::IfMetagenerationMatch>().value());
   }
   result.set_kms_key(request.GetOption<storage::KmsKeyName>().value_or(""));
+  if (request.GetOption<storage::DeleteSourceObjects>().value_or(false)) {
+    result.set_delete_source_objects(true);
+  }
   return result;
 }
 
diff --git a/google/cloud/storage/internal/object_requests.cc b/google/cloud/storage/internal/object_requests.cc
index 39d2c968cf271..5ca59b9060e2d 100644
--- a/google/cloud/storage/internal/object_requests.cc
+++ b/google/cloud/storage/internal/object_requests.cc
@@ -379,6 +379,9 @@ std::string ComposeObjectRequest::JsonPayload() const {
     source_object_list.emplace_back(std::move(source_object_json));
   }
   compose_object_payload_json["sourceObjects"] = source_object_list;
+  if (GetOption<DeleteSourceObjects>().value_or(false)) {
+    compose_object_payload_json["deleteSourceObjects"] = true;
+  }
 
   return compose_object_payload_json.dump();
 }
diff --git a/google/cloud/storage/internal/object_requests.h b/google/cloud/storage/internal/object_requests.h
index 6329b76eda038..6fd14a86b841b 100644
--- a/google/cloud/storage/internal/object_requests.h
+++ b/google/cloud/storage/internal/object_requests.h
@@ -266,10 +266,10 @@ std::ostream& operator<<(std::ostream& os, UpdateObjectRequest const& r);
  * Represents a request to the `Objects: compose` API.
  */
 class ComposeObjectRequest
-    : public GenericObjectRequest<ComposeObjectRequest, EncryptionKey,
-                                  DestinationPredefinedAcl, KmsKeyName,
-                                  IfGenerationMatch, IfMetagenerationMatch,
-                                  UserProject, WithObjectMetadata> {
+    : public GenericObjectRequest<
+          ComposeObjectRequest, EncryptionKey, DestinationPredefinedAcl,
+          KmsKeyName, IfGenerationMatch, IfMetagenerationMatch, UserProject,
+          WithObjectMetadata, DeleteSourceObjects> {
  public:
   ComposeObjectRequest() = default;
   explicit ComposeObjectRequest(std::string bucket_name,
diff --git a/google/cloud/storage/tests/object_rewrite_integration_test.cc b/google/cloud/storage/tests/object_rewrite_integration_test.cc
index ce4d49c26eac1..4519ebab467fb 100644
--- a/google/cloud/storage/tests/object_rewrite_integration_test.cc
+++ b/google/cloud/storage/tests/object_rewrite_integration_test.cc
@@ -293,6 +293,82 @@ TEST_F(ObjectRewriteIntegrationTest, ComposeSimple) {
   EXPECT_EQ(meta->size() * 2, composed_meta->size());
 }
 
+TEST_F(ObjectRewriteIntegrationTest, ComposeDeleteSourceObjectsFalse) {
+  auto client = MakeIntegrationTestClient();
+
+  auto object_name1 = MakeRandomObjectName();
+  StatusOr<ObjectMetadata> meta1 = client.InsertObject(
+      bucket_name_, object_name1, LoremIpsum(), IfGenerationMatch(0));
+  ASSERT_STATUS_OK(meta1);
+  ScheduleForDelete(*meta1);
+
+  auto object_name2 = MakeRandomObjectName();
+  StatusOr<ObjectMetadata> meta2 = client.InsertObject(
+      bucket_name_, object_name2, LoremIpsum(), IfGenerationMatch(0));
+  ASSERT_STATUS_OK(meta2);
+  ScheduleForDelete(*meta2);
+
+  auto composed_object_name = MakeRandomObjectName();
+  std::vector<ComposeSourceObject> source_objects = {{object_name1, {}, {}},
+                                                     {object_name2, {}, {}}};
+
+  StatusOr<ObjectMetadata> composed_meta = client.ComposeObject(
+      bucket_name_, source_objects, composed_object_name,
+      WithObjectMetadata(ObjectMetadata().set_content_type("plain/text")),
+      DeleteSourceObjects(false));
+  ASSERT_STATUS_OK(composed_meta);
+  ScheduleForDelete(*composed_meta);
+
+  EXPECT_EQ(meta1->size() + meta2->size(), composed_meta->size());
+
+  auto check1 = client.GetObjectMetadata(bucket_name_, object_name1);
+  EXPECT_STATUS_OK(check1) << "Source object 1 (" << object_name1
+                           << ") should NOT have been deleted.";
+
+  auto check2 = client.GetObjectMetadata(bucket_name_, object_name2);
+  EXPECT_STATUS_OK(check2) << "Source object 2 (" << object_name2
+                           << ") should NOT have been deleted.";
+}
+
+TEST_F(ObjectRewriteIntegrationTest, ComposeDeleteSourceObjectsTrue) {
+  auto client = MakeIntegrationTestClient();
+
+  auto object_name1 = MakeRandomObjectName();
+  StatusOr<ObjectMetadata> meta1 = client.InsertObject(
+      bucket_name_, object_name1, LoremIpsum(), IfGenerationMatch(0));
+  ASSERT_STATUS_OK(meta1);
+  ScheduleForDelete(*meta1);
+
+  auto object_name2 = MakeRandomObjectName();
+  StatusOr<ObjectMetadata> meta2 = client.InsertObject(
+      bucket_name_, object_name2, LoremIpsum(), IfGenerationMatch(0));
+  ASSERT_STATUS_OK(meta2);
+  ScheduleForDelete(*meta2);
+
+  auto composed_object_name = MakeRandomObjectName();
+  std::vector<ComposeSourceObject> source_objects = {{object_name1, {}, {}},
+                                                     {object_name2, {}, {}}};
+
+  StatusOr<ObjectMetadata> composed_meta = client.ComposeObject(
+      bucket_name_, source_objects, composed_object_name,
+      WithObjectMetadata(ObjectMetadata().set_content_type("plain/text")),
+      DeleteSourceObjects(true));
+  ASSERT_STATUS_OK(composed_meta);
+  ScheduleForDelete(*composed_meta);
+
+  EXPECT_EQ(meta1->size() + meta2->size(), composed_meta->size());
+
+  auto check1 = client.GetObjectMetadata(bucket_name_, object_name1);
+  EXPECT_FALSE(check1.ok());
+  EXPECT_EQ(StatusCode::kNotFound, check1.status().code())
+      << "Source object 1 (" << object_name1 << ") should have been deleted.";
+
+  auto check2 = client.GetObjectMetadata(bucket_name_, object_name2);
+  EXPECT_FALSE(check2.ok());
+  EXPECT_EQ(StatusCode::kNotFound, check2.status().code())
+      << "Source object 2 (" << object_name2 << ") should have been deleted.";
+}
+
 TEST_F(ObjectRewriteIntegrationTest, ComposedUsingEncryptedObject) {
   // TODO(#14385) - the emulator does not support this feature for gRPC.
   if (UsingEmulator() && UsingGrpc()) GTEST_SKIP();
diff --git a/google/cloud/storage/well_known_parameters.h b/google/cloud/storage/well_known_parameters.h
index b924322ea0c14..9eb8774ca726c 100644
--- a/google/cloud/storage/well_known_parameters.h
+++ b/google/cloud/storage/well_known_parameters.h
@@ -605,6 +605,17 @@ struct ReturnPartialSuccess
   }
 };
 
+/**
+ * Controls if the source objects should be deleted after a successful compose.
+ */
+struct DeleteSourceObjects
+    : public internal::WellKnownParameter<DeleteSourceObjects, bool> {
+  using WellKnownParameter<DeleteSourceObjects, bool>::WellKnownParameter;
+  static char const* well_known_parameter_name() {
+    return "deleteSourceObjects";
+  }
+};
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage
 }  // namespace cloud