[kafka] Copy cluster option to subscription (#3919)

Summary:
Make sure to always copy all kafka cluster additional options
to the subscription. Mainly the group.id.

Author: muhamadazmy

crates/types/src/schema/metadata/updater/mod.rs

@@ -934,12 +934,7 @@ impl SchemaUpdater {
 
         let subscription = Configuration::pinned()
             .ingress
-            .validate_subscription(Subscription::new(
-                id,
-                source,
-                sink,
-                metadata.unwrap_or_default(),
-            ))
+            .create_kafka_subscription(id, source, sink, metadata.unwrap_or_default())
             .map_err(|e| SchemaError::Subscription(SubscriptionError::Validation(e.into())))?;
 
         self.schema.subscriptions.insert(id, subscription);
@@ -1243,55 +1238,49 @@ pub struct ValidationError {
 }
 
 impl IngressOptions {
-    fn validate_subscription(
+    fn create_kafka_subscription(
         &self,
-        mut subscription: Subscription,
+        id: SubscriptionId,
+        source: Source,
+        sink: Sink,
+        mut metadata: HashMap<String, String>,
     ) -> Result<Subscription, ValidationError> {
         // Retrieve the cluster option and merge them with subscription metadata
-        let Source::Kafka { cluster, .. } = subscription.source();
+        let Source::Kafka { cluster, .. } = &source;
         let cluster_options = &self.get_kafka_cluster(cluster).ok_or(ValidationError {
             name: "source",
             reason: "specified cluster in the source URI does not exist. Make sure it is defined in the KafkaOptions",
         })?.additional_options;
 
         if cluster_options.contains_key("enable.auto.commit")
-            || subscription.metadata().contains_key("enable.auto.commit")
+            || metadata.contains_key("enable.auto.commit")
         {
             warn!(
                 "The configuration option enable.auto.commit should not be set and it will be ignored."
             );
         }
         if cluster_options.contains_key("enable.auto.offset.store")
-            || subscription
-                .metadata()
-                .contains_key("enable.auto.offset.store")
+            || metadata.contains_key("enable.auto.offset.store")
         {
             warn!(
                 "The configuration option enable.auto.offset.store should not be set and it will be ignored."
             );
         }
 
-        // Set the group.id if unset
-        if !(cluster_options.contains_key("group.id")
-            || subscription.metadata().contains_key("group.id"))
-        {
-            let group_id = subscription.id().to_string();
+        let group_id = metadata
+            .get("group.id")
+            .or_else(|| cluster_options.get("group.id"))
+            .cloned()
+            .unwrap_or_else(|| id.to_string());
 
-            subscription
-                .metadata_mut()
-                .insert("group.id".to_string(), group_id);
-        }
+        metadata.insert("group.id".into(), group_id);
 
         // Set client.id if unset
-        if !(cluster_options.contains_key("client.id")
-            || subscription.metadata().contains_key("client.id"))
-        {
-            subscription
-                .metadata_mut()
-                .insert("client.id".to_string(), "restate".to_string());
+        if !(cluster_options.contains_key("client.id") || metadata.contains_key("client.id")) {
+            metadata.insert("client.id".to_string(), "restate".to_string());
         }
 
-        Ok(subscription)
+        Ok(Subscription::new(id, source, sink, metadata))
     }
 }