feat: add kafka props enable.ssl.certificate.verification (#15073)

Signed-off-by: tabVersion <[email protected]>
Co-authored-by: xxchan <[email protected]>

Author: tabVersion

src/connector/src/common.rs

@@ -259,6 +259,10 @@ pub struct RdKafkaPropertiesCommon {
     #[serde(rename = "properties.client.id")]
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub client_id: Option<String>,
+
+    #[serde(rename = "properties.enable.ssl.certificate.verification")]
+    #[serde_as(as = "Option<DisplayFromStr>")]
+    pub enable_ssl_certificate_verification: Option<bool>,
 }
 
 impl RdKafkaPropertiesCommon {
@@ -275,6 +279,9 @@ impl RdKafkaPropertiesCommon {
         if let Some(v) = self.client_id.as_ref() {
             c.set("client.id", v);
         }
+        if let Some(v) = self.enable_ssl_certificate_verification {
+            c.set("enable.ssl.certificate.verification", v.to_string());
+        }
     }
 }
 

src/connector/with_options_sink.yaml

@@ -252,6 +252,9 @@ KafkaConfig:
     field_type: String
     comments: Client identifier
     required: false
+  - name: properties.enable.ssl.certificate.verification
+    field_type: bool
+    required: false
   - name: properties.allow.auto.create.topics
     field_type: bool
     comments: Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics.

src/connector/with_options_source.yaml

@@ -143,6 +143,9 @@ KafkaProperties:
     field_type: String
     comments: Client identifier
     required: false
+  - name: properties.enable.ssl.certificate.verification
+    field_type: bool
+    required: false
   - name: properties.queued.min.messages
     field_type: usize
     comments: Minimum number of messages per topic+partition librdkafka tries to maintain in the local  consumer queue.