This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 12f053b2eb2 Keep polling DataFusion pipeline state when the run is not
visible yet (#72406)
12f053b2eb2 is described below
commit 12f053b2eb2aed7a7dd76d4e83c2d74cea8e22f3
Author: Kunal <[email protected]>
AuthorDate: Sun Sep 13 19:22:48 2026 +0530
Keep polling DataFusion pipeline state when the run is not visible yet
(#72406)
#60688 changed `_check_response_status_and_data` to raise
`requests.exceptions.HTTPError` on a 404 instead of
`AirflowNotFoundException`, but the polling loop in
`DataFusionHook.wait_for_pipeline_state` was changed to catch `KeyError`
instead of `AirflowException`. The two no longer line up, so the 404 that
CDAP returns while a run is still being registered escapes the loop and
fails `CloudDataFusionStartPipelineOperator` immediately.
That 404 tolerance was added in #10031 for exactly this reason: right after
a pipeline is started, the run is not yet visible in the system.
Catch `HTTPError` alongside `KeyError` so the loop keeps polling. Only the
404 branch of `_check_response_status_and_data` raises `HTTPError` (any
other non-200 raises `RequestException`), so this stays narrow: real
failures still propagate.
---
.../providers/google/cloud/hooks/datafusion.py | 7 +++--
.../unit/google/cloud/hooks/test_datafusion.py | 33 ++++++++++++++++++++++
2 files changed, 38 insertions(+), 2 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
b/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
index a00331d3b1a..0da08fe4ea0 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
@@ -133,8 +133,11 @@ class DataFusionHook(GoogleBaseHook):
namespace=namespace,
)
current_state = workflow["status"]
- except KeyError:
- pass # Because the pipeline may not be visible in system yet
+ except (HTTPError, KeyError):
+ # A 404 is raised as HTTPError by
_check_response_status_and_data, and a
+ # missing "status" key raises KeyError. Both mean the run is
not visible
+ # in the system yet, so keep polling instead of failing the
task.
+ pass
if current_state in success_states:
return
if current_state in failure_states:
diff --git a/providers/google/tests/unit/google/cloud/hooks/test_datafusion.py
b/providers/google/tests/unit/google/cloud/hooks/test_datafusion.py
index ab39d1bf712..ce4b24bc1b6 100644
--- a/providers/google/tests/unit/google/cloud/hooks/test_datafusion.py
+++ b/providers/google/tests/unit/google/cloud/hooks/test_datafusion.py
@@ -588,6 +588,39 @@ class TestDataFusionHook:
method="GET",
)
+ @pytest.mark.parametrize(
+ "error",
+ [
+ pytest.param(HTTPError("Retrieving a pipeline state failed with
code 404"), id="404"),
+ pytest.param(KeyError("status"), id="missing-status"),
+ ],
+ )
+ @mock.patch(HOOK_STR.format("time.sleep"))
+ @mock.patch(HOOK_STR.format("DataFusionHook.get_pipeline_workflow"))
+ def test_wait_for_pipeline_state_keeps_polling_until_pipeline_is_visible(
+ self, mock_get_pipeline_workflow, mock_sleep, error, hook
+ ):
+ """A run that is not visible in the system yet must not fail the
task."""
+ mock_get_pipeline_workflow.side_effect = [error, {"status":
"COMPLETED"}]
+
+ hook.wait_for_pipeline_state(
+ pipeline_name=PIPELINE_NAME, pipeline_id=PIPELINE_ID,
instance_url=INSTANCE_URL
+ )
+
+ assert mock_get_pipeline_workflow.call_count == 2
+
+ @mock.patch(HOOK_STR.format("time.sleep"))
+ @mock.patch(HOOK_STR.format("DataFusionHook.get_pipeline_workflow"))
+ def test_wait_for_pipeline_state_raises_on_failure_state(
+ self, mock_get_pipeline_workflow, mock_sleep, hook
+ ):
+ mock_get_pipeline_workflow.return_value = {"status": "FAILED"}
+
+ with pytest.raises(ValueError, match=r"Pipeline shrubberyPipeline
state FAILED is not one of"):
+ hook.wait_for_pipeline_state(
+ pipeline_name=PIPELINE_NAME, pipeline_id=PIPELINE_ID,
instance_url=INSTANCE_URL
+ )
+
@pytest.mark.parametrize(
("pipeline_type", "expected_program_type"),
[
