remove private_link_targets and private_link_endpoint from struct

src/connector/src/common.rs

@@ -223,13 +223,7 @@ pub struct KafkaCommon {
 #[serde_as]
 #[derive(Debug, Clone, Deserialize, WithOptions)]
 pub struct KafkaPrivateLinkCommon {
-    // These two fields are only used in frontend to generate `broker_rewrite_map`
-    #[serde(rename = "privatelink.targets")]
-    pub private_link_targets: Option<String>,
-    #[serde(rename = "privatelink.endpoint")]
-    pub private_link_endpoint: Option<String>,
-
-    /// This is generated from `private_link_targets` and `private_link_endpoint`, instead of given by users.
+    /// This is generated from `private_link_targets` and `private_link_endpoint` in frontend, instead of given by users.
     #[serde(rename = "broker.rewrite.endpoints")]
     #[serde_as(as = "Option<JsonString>")]
     pub broker_rewrite_map: Option<HashMap<String, String>>,

src/connector/src/sink/kafka.rs

@@ -661,8 +661,6 @@ mod test {
             "properties.retry.max".to_string() => "20".to_string(),
             "properties.retry.interval".to_string() => "500ms".to_string(),
             // PrivateLink
-            "privatelink.targets".to_string() => "[{\"port\": 9292}]".to_string(),
-            "privatelink.endpoint".to_string() => "10.0.0.1".to_string(),
             "broker.rewrite.endpoints".to_string() => "{\"broker1\": \"10.0.0.1:8001\"}".to_string(),
         };
         let config = KafkaConfig::from_hashmap(properties).unwrap();
@@ -672,14 +670,6 @@ mod test {
         assert_eq!(config.retry_interval, Duration::from_millis(500));
 
         // PrivateLink fields
-        assert_eq!(
-            config.privatelink_common.private_link_endpoint,
-            Some("10.0.0.1".into())
-        );
-        assert_eq!(
-            config.privatelink_common.private_link_targets,
-            Some("[{\"port\": 9292}]".into())
-        );
         let hashmap: HashMap<String, String> = hashmap! {
             "broker1".to_string() => "10.0.0.1:8001".to_string()
         };

src/connector/src/source/kafka/mod.rs

@@ -214,8 +214,6 @@ mod test {
             "properties.enable.auto.commit".to_string() => "true".to_string(),
             "properties.fetch.queue.backoff.ms".to_string() => "114514".to_string(),
             // PrivateLink
-            "privatelink.targets".to_string() => "[{\"port\": 9292}]".to_string(),
-            "privatelink.endpoint".to_string() => "10.0.0.1".to_string(),
             "broker.rewrite.endpoints".to_string() => "{\"broker1\": \"10.0.0.1:8001\"}".to_string(),
         };
 
@@ -255,15 +253,6 @@ mod test {
             props.rdkafka_properties_consumer.fetch_queue_backoff_ms,
             Some(114514)
         );
-        assert_eq!(
-            props.privatelink_common.private_link_endpoint,
-            Some("10.0.0.1".into())
-        );
-        assert_eq!(
-            props.privatelink_common.private_link_targets,
-            Some("[{\"port\": 9292}]".into())
-        );
-
         let hashmap: HashMap<String, String> = hashmap! {
             "broker1".to_string() => "10.0.0.1:8001".to_string()
         };

src/connector/with_options_sink.yaml

@@ -302,15 +302,9 @@ KafkaConfig:
     comments: The maximum number of unacknowledged requests the client will send on a single connection before blocking.
     required: false
     default: '5'
-  - name: privatelink.targets
-    field_type: String
-    required: false
-  - name: privatelink.endpoint
-    field_type: String
-    required: false
   - name: broker.rewrite.endpoints
     field_type: HashMap<String,String>
-    comments: This is generated from `private_link_targets` and `private_link_endpoint`, instead of given by users.
+    comments: This is generated from `private_link_targets` and `private_link_endpoint` in frontend, instead of given by users.
     required: false
 KinesisSinkConfig:
   fields:

src/connector/with_options_source.yaml

@@ -166,15 +166,9 @@ KafkaProperties:
     field_type: bool
     comments: 'Automatically and periodically commit offsets in the background.  Note: setting this to false does not prevent the consumer from fetching previously committed start offsets.  To circumvent this behaviour set specific start offsets per partition in the call to assign().  default: true'
     required: false
-  - name: privatelink.targets
-    field_type: String
-    required: false
-  - name: privatelink.endpoint
-    field_type: String
-    required: false
   - name: broker.rewrite.endpoints
     field_type: HashMap<String,String>
-    comments: This is generated from `private_link_targets` and `private_link_endpoint`, instead of given by users.
+    comments: This is generated from `private_link_targets` and `private_link_endpoint` in frontend, instead of given by users.
     required: false
 KinesisProperties:
   fields: