Persist DataprocLink for workflow operators regardless of job status (#…

…26986) For DataprocInstantiateInlineWorkflowTemplateOperator and DataprocInstantiateWorkflowTemplateOperator, the dataproc link is available only for the jobs that have succeeded. Incase of Failure, the DataprocLink is not available
apache · Nov 16, 2022 · 0cb6450 · 0cb6450
1 parent 7cfa1be
commit 0cb6450
Show file tree

Hide file tree

Showing 2 changed files with 194 additions and 97 deletions.
diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py
@@ -1679,12 +1679,13 @@ def execute(self, context: Context):
  timeout=self.timeout,
  metadata=self.metadata,
  )
- operation.result()
- workflow_id = operation.operation.name.split("/")[-1]
+ self.workflow_id = operation.operation.name.split("/")[-1]
  DataprocLink.persist(
- context=context, task_instance=self, url=DATAPROC_WORKFLOW_LINK, resource=workflow_id
+ context=context, task_instance=self, url=DATAPROC_WORKFLOW_LINK, resource=self.workflow_id
  )
- self.log.info("Template instantiated.")
+ self.log.info("Template instantiated. Workflow Id : %s", self.workflow_id)
+ operation.result()
+ self.log.info("Workflow %s completed successfully", self.workflow_id)
 
 
 class DataprocInstantiateInlineWorkflowTemplateOperator(BaseOperator):
@@ -1770,12 +1771,13 @@ def execute(self, context: Context):
  timeout=self.timeout,
  metadata=self.metadata,
  )
- operation.result()
- workflow_id = operation.operation.name.split("/")[-1]
+ self.workflow_id = operation.operation.name.split("/")[-1]
  DataprocLink.persist(
- context=context, task_instance=self, url=DATAPROC_WORKFLOW_LINK, resource=workflow_id
+ context=context, task_instance=self, url=DATAPROC_WORKFLOW_LINK, resource=self.workflow_id
  )
- self.log.info("Template instantiated.")
+ self.log.info("Template instantiated. Workflow Id : %s", self.workflow_id)
+ operation.result()
+ self.log.info("Workflow %s completed successfully", self.workflow_id)
 
 
 class DataprocSubmitJobOperator(BaseOperator):