Adding a new workunit processor to check correctness of an aspect

metadata-ingestion/src/datahub/ingestion/api/source_helpers.py

@@ -78,6 +78,25 @@ def create_dataset_props_patch_builder(
     return patch_builder
 
 
+def check_mcp_correctness(mcp: MetadataChangeProposalClass):
+    logger.debug(f"Processing as MCP with urn: {mcp.entityUrn} and aspect: {mcp.aspectName}, change type: {mcp.changeType}")
+    logger.debug(f"{mcp.aspect}")
+
+
+def check_mcpw_correctness(mcp: MetadataChangeProposalWrapper):
+    logger.debug(f"Processing as MCP with urn: {mcp.entityUrn} and aspect: {mcp.aspectName}, change type: {mcp.changeType}")
+    logger.debug(f"{mcp.aspect}")
+
+
+def check_workunit_correctness(stream: Iterable[MetadataWorkUnit]) -> Iterable[MetadataWorkUnit]:
+    for wu in stream:
+        logger.debug(f"Checking correctnes for workunit: {wu.id}")
+        if isinstance(wu.metadata, MetadataChangeProposalClass):
+            check_mcp_correctness(wu.metadata)
+        elif isinstance(wu.metadata, MetadataChangeProposalWrapper):
+            check_mcpw_correctness(wu.metadata)
+
+
 def create_dataset_owners_patch_builder(
     dataset_urn: str,
     ownership: Ownership,

metadata-ingestion/src/datahub/ingestion/source/unity/source.py

@@ -42,7 +42,7 @@
 )
 from datahub.ingestion.api.source_helpers import (
     create_dataset_owners_patch_builder,
-    create_dataset_props_patch_builder,
+    create_dataset_props_patch_builder, check_workunit_correctness,
 )
 from datahub.ingestion.api.workunit import MetadataWorkUnit
 from datahub.ingestion.source.aws import s3_util
@@ -260,6 +260,7 @@ def get_workunit_processors(self) -> List[Optional[MetadataWorkUnitProcessor]]:
             StaleEntityRemovalHandler.create(
                 self, self.config, self.ctx
             ).workunit_processor,
+            check_workunit_correctness
         ]
 
     def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]: