feat(kafka-connect): add column-level lineage support for MongoDB source connector

MongoSourceConnector.extract_lineages() now calls _extract_fine_grained_lineage()
to produce column-level lineage when use_schema_resolver and
schema_resolver_finegrained_lineage are enabled. The base class method resolves
MongoDB collection schemas from DataHub and creates 1:1 field mappings.

Prerequisite: MongoDB collections must be ingested into DataHub before
Kafka Connect ingestion for schema resolution to work.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: treff7es

metadata-ingestion/src/datahub/ingestion/source/kafka_connect/source_connectors.py

@@ -2121,11 +2121,19 @@ def extract_lineages(self) -> List[KafkaConnectLineage]:
             if found:
                 table_name = get_dataset_name(found.group(1), found.group(2))
 
+                fine_grained = self._extract_fine_grained_lineage(
+                    source_dataset=table_name,
+                    source_platform=source_platform,
+                    target_dataset=topic,
+                    target_platform=KAFKA,
+                )
+
                 lineage = KafkaConnectLineage(
                     source_dataset=table_name,
                     source_platform=source_platform,
                     target_dataset=topic,
                     target_platform=KAFKA,
+                    fine_grained_lineages=fine_grained,
                 )
                 lineages.append(lineage)
         return lineages

metadata-ingestion/tests/unit/test_kafka_connect.py

@@ -604,6 +604,65 @@ def test_mongo_source_lineage_topic_parsing(self) -> None:
         assert lineages[0].source_dataset == "my-new-database.users"
         assert lineages[1].source_dataset == "-leading-hyphen._leading-underscore"
 
+    def test_mongo_source_fine_grained_lineage(self) -> None:
+        """Test MongoDB connector produces column-level lineage when schema resolver is available."""
+        config = KafkaConnectSourceConfig(
+            connect_uri="http://test:8083",
+            cluster_name="test",
+            use_schema_resolver=True,
+            schema_resolver_finegrained_lineage=True,
+        )
+        report = KafkaConnectSourceReport()
+
+        manifest = ConnectorManifest(
+            name="mongo-source-connector",
+            type="source",
+            config={
+                "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
+                "topic.prefix": "prod.mongo",
+            },
+            tasks=[],
+            topic_names=["prod.mongo.mydb.users"],
+        )
+
+        mock_resolver = Mock()
+        mock_resolver.resolve_table.return_value = (
+            "urn:li:dataset:(urn:li:dataPlatform:mongodb,mydb.users,PROD)",
+            {"_id": "string", "name": "string", "email": "string"},
+        )
+
+        connector = MongoSourceConnector(
+            connector_manifest=manifest,
+            config=config,
+            report=report,
+            schema_resolver=mock_resolver,
+        )
+
+        lineages = connector.extract_lineages()
+
+        assert len(lineages) == 1
+        assert lineages[0].source_dataset == "mydb.users"
+        assert lineages[0].target_dataset == "prod.mongo.mydb.users"
+        assert lineages[0].fine_grained_lineages is not None
+        assert len(lineages[0].fine_grained_lineages) == 3
+
+    def test_mongo_source_no_fine_grained_without_schema_resolver(self) -> None:
+        """Test MongoDB connector skips column-level lineage when schema resolver is disabled."""
+        connector_config: Dict[str, str] = {
+            "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
+            "topic.prefix": "prod.mongo.avro",
+        }
+
+        manifest = self.create_mock_manifest(connector_config)
+        config = create_mock_kafka_connect_config()
+        report = Mock(spec=KafkaConnectSourceReport)
+
+        connector = MongoSourceConnector(manifest, config, report)
+        lineages = connector.extract_lineages()
+
+        assert len(lineages) == 2
+        assert all(lin.fine_grained_lineages is None for lin in lineages)
+
 
 class TestConfluentCloudConnectors:
     """Test Confluent Cloud connector compatibility with Platform connectors."""