From 7c394a40c355d2923341673ce8d361cfc28ed7dd Mon Sep 17 00:00:00 2001 From: StrikeW Date: Fri, 2 Feb 2024 16:53:10 +0800 Subject: [PATCH] remove private_link_targets and private_link_endpoint from struct --- src/connector/src/common.rs | 8 +------- src/connector/src/sink/kafka.rs | 10 ---------- src/connector/src/source/kafka/mod.rs | 11 ----------- src/connector/with_options_sink.yaml | 8 +------- src/connector/with_options_source.yaml | 8 +------- 5 files changed, 3 insertions(+), 42 deletions(-) diff --git a/src/connector/src/common.rs b/src/connector/src/common.rs index bdb7f087d3659..25179bb209866 100644 --- a/src/connector/src/common.rs +++ b/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, - #[serde(rename = "privatelink.endpoint")] - pub private_link_endpoint: Option, - - /// 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")] pub broker_rewrite_map: Option>, diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs index 54e8475c13734..25681125f9069 100644 --- a/src/connector/src/sink/kafka.rs +++ b/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 = hashmap! { "broker1".to_string() => "10.0.0.1:8001".to_string() }; diff --git a/src/connector/src/source/kafka/mod.rs b/src/connector/src/source/kafka/mod.rs index 8809b589d4bd0..52d410a8ee717 100644 --- a/src/connector/src/source/kafka/mod.rs +++ b/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 = hashmap! { "broker1".to_string() => "10.0.0.1:8001".to_string() }; diff --git a/src/connector/with_options_sink.yaml b/src/connector/with_options_sink.yaml index 3dd0cf79530c3..3b011a8b71288 100644 --- a/src/connector/with_options_sink.yaml +++ b/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 - 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: diff --git a/src/connector/with_options_source.yaml b/src/connector/with_options_source.yaml index a47f721f65e38..3ef323aa88574 100644 --- a/src/connector/with_options_source.yaml +++ b/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 - 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: