Fixes #31152: emit Snowflake Sink lineage in the KafkaConnect connector

ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/client.py

@@ -307,9 +307,6 @@ def get_connector_config(self, connector: str) -> Optional[dict]:  # noqa: UP045
         """
         Get the details of a single connector.
 
-        For Confluent Cloud, the API returns configs as an array of {config, value} objects.
-        For self-hosted Kafka Connect, it returns a flat config dictionary.
-
         Args:
             connector (str): The name of the connector.
         """
@@ -318,19 +315,8 @@ def get_connector_config(self, connector: str) -> Optional[dict]:  # noqa: UP045
             if not result:
                 return None
 
-            # Check if this is Confluent Cloud format (array of {config, value})
-            if self.is_confluent_cloud and "configs" in result:
-                # Transform Confluent Cloud format: [{config: "key", value: "val"}] -> {key: val}
-                configs_array = result.get("configs", [])
-                if isinstance(configs_array, list):
-                    config_dict = {
-                        item["config"]: item["value"]
-                        for item in configs_array
-                        if isinstance(item, dict) and "config" in item and "value" in item
-                    }
-                    return config_dict or None
-
-            # Standard self-hosted Kafka Connect format
+            # Confluent Cloud and self-hosted Connect both return a flat config map here
+            # (verified against the Connect v1 API on 2026-08-05).
             return result.get("config")
 
         except Exception as exc:

ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/constants.py

@@ -35,13 +35,25 @@ class ConnectorConfigKeys:
         "snowflake.topic2table.map",  # Snowflake Sink: Critical mapping (e.g., "topicA:tableA, topicB:tableB")
     ]
 
+    # Both key forms the Snowflake sink accepts, most specific first. Spliced into the
+    # generic lists below and read by SnowflakeSinkResolver, so the dedicated resolver
+    # cannot recognise fewer keys than the generic key-list search it replaces.
+    SNOWFLAKE_DATABASE_KEYS = [  # noqa: RUF012
+        "snowflake.database.name",  # Snowflake: The target database
+        "snowflake.database",  # Snowflake: Variation
+    ]
+
+    SNOWFLAKE_SCHEMA_KEYS = [  # noqa: RUF012
+        "snowflake.schema.name",  # Snowflake: The Schema (e.g. "PUBLIC")
+        "snowflake.schema",  # Snowflake variation
+    ]
+
     DATABASE_KEYS = [  # noqa: RUF012
         "database",  # Generic: Common in simple JDBC configs
         "db.name",  # Generic: Common variation
         "database.dbname",  # PostgreSQL/JDBC: The physical database name
         "topic.prefix",  # Debezium: The "Logical Server Name".
-        "snowflake.database.name",  # Snowflake: The target database
-        "snowflake.database",  # Snowflake: Variation
+        *SNOWFLAKE_DATABASE_KEYS,
         "defaultDataset",  # BigQuery: The Dataset (Equivalent to a Database/Schema)
         "mongodb.database",  # MongoDB: The specific database to watch/write to
         "cassandra.keyspace",  # Cassandra: Keyspace is the Cassandra equivalent of a Database
@@ -55,8 +67,7 @@ class ConnectorConfigKeys:
     ]
 
     SCHEMA_KEYS = [  # noqa: RUF012
-        "snowflake.schema.name",  # Snowflake: The Schema (e.g. "PUBLIC")
-        "snowflake.schema",  # Snowflake variation
+        *SNOWFLAKE_SCHEMA_KEYS,
         "schema.name",  # Generic JDBC: Schema namespace
     ]
 
@@ -115,6 +126,9 @@ class ConnectorConfigKeys:
     "MongoDbCdcSource": "MongoDB",
     "OracleCdcSource": "Oracle",
     "Db2CdcSource": "Db2",
+    # Confluent Cloud reports the short plugin name; self-managed Connect reports the Java class.
+    "SnowflakeSink": "Snowflake",
+    "SnowflakeSinkConnector": "Snowflake",
 }
 
 # Map service types to hostname config keys
@@ -124,6 +138,20 @@ class ConnectorConfigKeys:
     "Mssql": ["database.hostname"],
     "MongoDB": ["mongodb.connection.uri", "connection.uri"],
     "Oracle": ["database.hostname"],
+    "Snowflake": ["snowflake.url.name"],
+}
+
+# Service connection attributes probed, in order, for the host identifying a service.
+# Most connections expose hostPort or host; Snowflake exposes neither and identifies
+# the deployment by `account`.
+SERVICE_CONNECTION_HOST_ATTRIBUTES = ["hostPort", "host", "account"]
+
+# Domain suffixes a connector may append to the host stored on the service connection.
+# Confluent reports "<account>.snowflakecomputing.com" for snowflake.url.name while the
+# OpenMetadata Snowflake service stores the bare "<account>", so the suffix must not
+# defeat the comparison. Values must be lowercase: hosts are lowercased before matching.
+SERVICE_TYPE_HOST_DOMAIN_SUFFIXES = {
+    "Snowflake": [".snowflakecomputing.com"],
 }
 
 # Map service types to broker/endpoint config keys for messaging services