add matching logic for domain ingestion

FundingCircle · May 10, 2024 · 21b96f3 · 21b96f3
1 parent 35d879d
commit 21b96f3
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/metadata-ingestion/src/datahub/ingestion/source/kafka.py b/metadata-ingestion/src/datahub/ingestion/source/kafka.py
@@ -421,6 +421,10 @@ def _extract_record(
                         for tag_association in meta_tags_aspect.tags
                     ]
 
+                meta_domain_aspect = meta_aspects.get(Constants.ADD_DOMAIN_OPERATION)
+                if meta_domain_aspect:
+                    dataset_snapshot.aspects.append(meta_domain_aspect)
+
             if all_tags:
                 dataset_snapshot.aspects.append(
                     mce_builder.make_global_tag_aspect_with_tag_list(all_tags)

diff --git a/metadata-ingestion/src/datahub/utilities/mapping.py b/metadata-ingestion/src/datahub/utilities/mapping.py
@@ -381,7 +381,9 @@ def get_operation_value(
                 if term.strip()
             ]
         elif operation_type == Constants.ADD_DOMAIN_OPERATION:
-            return mce_builder.make_domain_urn(operation_config[Constants.DOMAIN])
+            domain = operation_config[Constants.DOMAIN]
+            domain = _insert_match_value(domain, _get_best_match(match, "domain"))
+            return mce_builder.make_domain_urn(domain)
         return None
 
     def sanitize_owner_ids(self, owner_id: str) -> str: