Fix#10377: service connection not overwritten as expected (#10445)

ingestion/src/metadata/ingestion/api/workflow.py

@@ -304,7 +304,10 @@ class Workflow(WorkflowStatusMixin):
         :param service_type: source workflow service type
         :return:
         """
-        if not self.config.source.serviceConnection:
+        if (
+            not self.config.source.serviceConnection
+            and not self.metadata.config.forceEntityOverwriting
+        ):
             service_name = self.config.source.serviceName
             try:
                 service: ServiceWithConnectionType = cast(

ingestion/src/metadata/profiler/api/workflow.py

@@ -488,7 +488,10 @@ class ProfilerWorkflow(WorkflowStatusMixin):
         service_type: ServiceType = get_service_type_from_source_type(
             self.config.source.type
         )
-        if not self.config.source.serviceConnection:
+        if (
+            not self.config.source.serviceConnection
+            and not self.metadata.config.forceEntityOverwriting
+        ):
             service_name = self.config.source.serviceName
             try:
                 service: ServiceWithConnectionType = cast(

ingestion/src/metadata/test_suite/api/workflow.py

@@ -82,7 +82,6 @@ class TestSuiteWorkflow(WorkflowStatusMixin):
 
         Attributes:
             config: OM workflow configuration object
-            source_config: TestSuitePipeline object
         """
         self.config = config
 
@@ -182,14 +181,14 @@ class TestSuiteWorkflow(WorkflowStatusMixin):
                     service_connection_config.catalog = entity_fqn.split(".")[1]
             return service_connection_config
 
-        logger.error(f"Could not retrive connection details for entity {entity_link}")
+        logger.error(f"Could not retrieve connection details for entity {entity_link}")
         raise ValueError()
 
     def _get_table_entity_from_test_case(self, entity_fqn: str):
         """given an entityLink return the table entity
 
         Args:
-            entity_link: entity link for the test case
+            entity_fqn: entity fqn for the test case
         """
         return self.metadata.get_by_name(
             entity=Table,
@@ -298,8 +297,7 @@ class TestSuiteWorkflow(WorkflowStatusMixin):
         self,
     ) -> List[TestSuite]:
         """
-        Fro the CLI workflow we'll have n testSuite in the
+        For the CLI workflow we'll have n testSuite in the processor.config.testSuites
-        processor.config.testSuites
         """
         test_suite_entities = []
         test_suites = self.processor_config.testSuites or []

ingestion/tests/integration/workflow/test_workflow.py

@@ -86,6 +86,19 @@ class WorkflowTest(TestCase):
         except ConfigurationError:
             self.assertRaises(ConfigurationError)
 
+    def test_fail_no_service_connection_and_overwrite(self):
+        current_dir = pathlib.Path(__file__).resolve().parent
+        config_file = current_dir.joinpath("mysql_test.yaml")
+        workflow_config = load_config_file(config_file)
+
+        del workflow_config["source"]["serviceConnection"]
+        workflow_config["workflowConfig"]["openMetadataServerConfig"][
+            "forceEntityOverwriting"
+        ] = True
+
+        with self.assertRaises(AttributeError):
+            Workflow.create(workflow_config)
+
     def test_debug_not_show_authorization_headers(self):
         current_dir = pathlib.Path(__file__).resolve().parent
         config_file = current_dir.joinpath("mysql_test.yaml")

openmetadata-docs/content/connectors/ingestion/deployment/index.md

@@ -213,6 +213,7 @@ And we receive the created Ingestion Pipeline Entity back:
     "enableVersionValidation": true,
     "includeDatabaseServices": true,
     "includePipelineServices": true,
+    "forceEntityOverwriting": false,
     "limitRecords": 1000,
     "supportsMetadataExtraction": true
   },

openmetadata-docs/content/main-concepts/metadata-standard/schemas/entity/services/connections/metadata/openMetadataConnection.md

@@ -29,6 +29,7 @@ slug: /main-concepts/metadata-standard/schemas/entity/services/connections/metad
 - **`enableVersionValidation`** *(boolean)*: Validate Openmetadata Server & Client Version. Default: `True`.
 - **`includeDatabaseServices`** *(boolean)*: Include Database Services for Indexing. Default: `True`.
 - **`includePipelineServices`** *(boolean)*: Include Pipeline Services for Indexing. Default: `True`.
+- **`forceEntityOverwriting`** *(boolean)*: 'Force the overwriting of any entity during the ingestion. Default: `False`.
 - **`limitRecords`** *(integer)*: Limit the number of records for Indexing. Default: `1000`.
 - **`supportsMetadataExtraction`**: Refer to *../connectionBasicType.json#/definitions/supportsMetadataExtraction*.
 ## Definitions