From 6456512fbdc3b0fc6e1d51a5c4f22ceb66ab2fe3 Mon Sep 17 00:00:00 2001
From: Matt Chamberlin <MChamberlin@users.noreply.github.com>
Date: Wed, 3 Jul 2024 10:47:28 -0400
Subject: [PATCH] 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>
---
 .../source/pipeline/openlineage/connection.py |  9 +++++++
 .../pipeline/openLineageConnection.json       | 11 +++++++-
 .../security/sasl/saslClientConfig.json       | 27 +++++++++++++++++++
 .../locales/en-US/Pipeline/OpenLineage.md     | 23 +++++++++++++++-
 4 files changed, 68 insertions(+), 2 deletions(-)
 create mode 100644 openmetadata-spec/src/main/resources/json/schema/security/sasl/saslClientConfig.json

diff --git a/ingestion/src/metadata/ingestion/source/pipeline/openlineage/connection.py b/ingestion/src/metadata/ingestion/source/pipeline/openlineage/connection.py
index e1edfb839bd..74d488d7743 100644
--- a/ingestion/src/metadata/ingestion/source/pipeline/openlineage/connection.py
+++ b/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])
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/pipeline/openLineageConnection.json b/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/pipeline/openLineageConnection.json
index 9020e49aa5f..637c192fde7 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/pipeline/openLineageConnection.json
+++ b/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"
     }
diff --git a/openmetadata-spec/src/main/resources/json/schema/security/sasl/saslClientConfig.json b/openmetadata-spec/src/main/resources/json/schema/security/sasl/saslClientConfig.json
new file mode 100644
index 00000000000..51debf9197a
--- /dev/null
+++ b/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"
+    }
+  }
+}
diff --git a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Pipeline/OpenLineage.md b/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Pipeline/OpenLineage.md
index 8e10e30fe5d..ba4fba89396 100644
--- a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Pipeline/OpenLineage.md
+++ b/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