feat: allow kafka source/sink config `ssl_endpoint_identification_alg…

…orithm` (#13990)

src/connector/src/common.rs

@@ -167,6 +167,9 @@ pub struct KafkaCommon {
     #[serde(rename = "properties.security.protocol")]
     security_protocol: Option<String>,
 
+    #[serde(rename = "properties.ssl.endpoint.identification.algorithm")]
+    ssl_endpoint_identification_algorithm: Option<String>,
+
     // For the properties below, please refer to [librdkafka](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md) for more information.
     /// Path to CA certificate file for verifying the broker's key.
     #[serde(rename = "properties.ssl.ca.location")]
@@ -294,6 +297,15 @@ impl KafkaCommon {
         if let Some(ssl_key_password) = self.ssl_key_password.as_ref() {
             config.set("ssl.key.password", ssl_key_password);
         }
+        if let Some(ssl_endpoint_identification_algorithm) =
+            self.ssl_endpoint_identification_algorithm.as_ref()
+        {
+            // accept only `none` and `http` here, let the sdk do the check
+            config.set(
+                "ssl.endpoint.identification.algorithm",
+                ssl_endpoint_identification_algorithm,
+            );
+        }
 
         // SASL mechanism
         if let Some(sasl_mechanism) = self.sasl_mechanism.as_ref() {

src/connector/with_options_sink.yaml

@@ -170,6 +170,9 @@ KafkaConfig:
     field_type: String
     comments: Security protocol used for RisingWave to communicate with Kafka brokers. Could be  PLAINTEXT, SSL, SASL_PLAINTEXT or SASL_SSL.
     required: false
+  - name: properties.ssl.endpoint.identification.algorithm
+    field_type: String
+    required: false
   - name: properties.ssl.ca.location
     field_type: String
     comments: Path to CA certificate file for verifying the broker's key.

src/connector/with_options_source.yaml

@@ -43,6 +43,9 @@ KafkaProperties:
     field_type: String
     comments: Security protocol used for RisingWave to communicate with Kafka brokers. Could be  PLAINTEXT, SSL, SASL_PLAINTEXT or SASL_SSL.
     required: false
+  - name: properties.ssl.endpoint.identification.algorithm
+    field_type: String
+    required: false
   - name: properties.ssl.ca.location
     field_type: String
     comments: Path to CA certificate file for verifying the broker's key.