Repository navigation
Issue with CloudDataFusionStartPipelineOperator #10030
Description
I am using the CloudDataFusionStartPipelineOperator in Google cloud composer and success_states = ["COMPLETED"] parameter to wait for the pipeline to be successful. It works fine, but sometimes I receive this error:
[2020-07-28 08:16:46,390] {taskinstance.py:1059} ERROR - Retrieving a pipeline state failed with code 404
Traceback (most recent call last)
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 930, in _run_raw_tas
result = task_copy.execute(context=context
File "/usr/local/lib/airflow/airflow/providers/google/cloud/operators/datafusion.py", line 700, in execut
runtime_args=self.runtime_args
File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/datafusion.py", line 498, in start_pipelin
instance_url=instance_url
File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/datafusion.py", line 114, in wait_for_pipeline_stat
namespace=namespace
File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/datafusion.py", line 442, in _get_workflow_stat
f"Retrieving a pipeline state failed with code {response.status}
airflow.exceptions.AirflowException: Retrieving a pipeline state failed with code 404
But the Datafusion pipeline gets completed successfully.
It would be helpful if someone resolves this.
Reacted by Tomek Urbaszek
Activity
boring-cyborg commentedon Jul 28, 2020
Thanks for opening your first issue here! Be sure to follow the issue template!
Add try clause to DataFusionHook.wait_for_pipeline_state
turbaszek commentedon Jul 28, 2020
Thanks for reporting @akspow ! I think this may be related to the fact that DataFusion happens to have some lags (I experienced it when writing those operators, see the example DAG where we use dummy operator with sleep)
akspow commentedon Jul 29, 2020
Is there a quick fix for this?
turbaszek commentedon Jul 29, 2020
You can create custom operator using the logic from the existing one and add the change from #10031
Add try clause to DataFusionHook.wait_for_pipeline_state (#10031)