Fixes 16582: support SASL_SSL kafka auth for OpenLineage pipeline connector (#16597)

* support SASL_SSL kafka auth for OpenLineage pipeline connector

* update documentation

* use existing saslMechanismType schema

---------

Co-authored-by: Matthew Chamberlin <mchamberlin@ginkgobioworks.com>

ingestion/src/metadata/ingestion/source/pipeline/openlineage/connection.py

@@ -52,6 +52,15 @@ def get_connection(connection: OpenLineageConnection) -> KafkaConsumer:
                     "ssl.key.location": connection.sslConfig.root.sslKey,
                 }
             )
+        elif connection.securityProtocol.value == KafkaSecProtocol.SASL_SSL.value:
+            config.update(
+                {
+                    "security.protocol": connection.securityProtocol.value,
+                    "sasl.mechanism": connection.saslConfig.saslMechanism.value,
+                    "sasl.username": connection.saslConfig.saslUsername,
+                    "sasl.password": connection.saslConfig.saslPassword,
+                }
+            )
 
         kafka_consumer = KafkaConsumer(config)
         kafka_consumer.subscribe([connection.topicName])

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/pipeline/openLineageConnection.json

@@ -73,7 +73,8 @@
       "type": "string",
       "enum": [
         "PLAINTEXT",
-        "SSL"
+        "SSL",
+        "SASL_SSL"
       ],
       "javaEnums": [
         {
@@ -81,6 +82,9 @@
         },
         {
           "name": "SSL"
+        },
+        {
+          "name": "SASL_SSL"
         }
       ]
     },
@@ -89,6 +93,11 @@
       "description": "SSL Configuration details.",
       "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/sslConfig"
     },
+    "saslConfig": {
+      "title": "SASL",
+      "description": "SASL Configuration details.",
+      "$ref": "../../../../security/sasl/saslClientConfig.json"
+    },
     "supportsMetadataExtraction": {
       "$ref": "../connectionBasicType.json#/definitions/supportsMetadataExtraction"
     }

openmetadata-spec/src/main/resources/json/schema/security/sasl/saslClientConfig.json

@@ -0,0 +1,27 @@
+{
+  "$id": "https://open-metadata.org/schema/security/sasl/SASLClientConfig.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "SASL Client Config",
+  "description": "SASL client configuration.",
+  "type": "object",
+  "javaType": "org.openmetadata.schema.security.sasl.SASLClientConfig",
+  "additionalProperties": false,
+  "properties": {
+    "saslMechanism": {
+      "title": "SASL Mechanism",
+      "description": "SASL security mechanism",
+      "$ref": "../../entity/services/connections/messaging/saslMechanismType.json",
+      "default": "PLAIN"
+    },
+    "saslUsername": {
+      "title": "SASL Username",
+      "description": "The SASL authentication username.",
+      "type": "string"
+    },
+    "saslPassword": {
+      "title": "SASL Password",
+      "description": "The SASL authentication password.",
+      "type": "string"
+    }
+  }
+}

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Pipeline/OpenLineage.md

@@ -59,7 +59,28 @@ $$section
 ### Kafka securityProtocol $(id="securityProtocol")
 Kafka Security protocol config.
 
-This should be specified as `PLAINTEXT` or `SSL` .
+This should be specified as `PLAINTEXT`, `SSL`, or `SASL_SSL` .
+$$
+
+$$section
+### Kafka SASL mechanism $(id="saslMechanism")
+When Kafka security protocol is set to `SASL_SSL` then the SASL mechanism is needed.
+
+This should be specified as `PLAIN` .
+$$
+
+$$section
+### Kafka SASL username $(id="saslUsername")
+When Kafka security protocol is set to `SASL_SSL` then the SASL username is needed.
+
+This should be specified as a username or API key string .
+$$
+
+$$section
+### Kafka SASL password $(id="saslPassword")
+When Kafka security protocol is set to `SASL_SSL` then the SASL password is needed.
+
+This should be specified as a password or API secret string .
 $$
 
 $$section