Fix Data Fusion polling after transient 404 responses

[VladaZakharova]

Upstream refactoring PR #60688 (released in apache-airflow-providers-google==20.0.0) moved pipeline existence/state checks into DataFusionHook and DataFusionAsyncHook, breaking transient 404 handling across all three polling execution modes:

1. Synchronous Mode (CloudDataFusionStartPipelineOperator with asynchronous=False, deferrable=False):

DataFusionHook._check_response_status_and_data() raises requests.exceptions.HTTPError on HTTP 404, but PR #60688 changed DataFusionHook.wait_for_pipeline_state() from except Exception: to except KeyError:. (was fixed in #72406)

2. Sensor Mode (CloudDataFusionPipelineStateSensor.poke in airflow/providers/google/cloud/sensors/datafusion.py):

Prior to PR #60688, DataFusionHook.get_pipeline_workflow() raised AirflowNotFoundException on HTTP 404 and AirflowException on other HTTP errors.
CloudDataFusionPipelineStateSensor.poke() was written to catch except AirflowNotFoundException: (and return False to wait for the next poke interval) and except AirflowException:.
PR #60688 changed get_pipeline_workflow() to call self._check_response_status_and_data(), which now raises requests.exceptions.HTTPError (on 404) and requests.exceptions.RequestException (on non-200), but sensors/datafusion.py was never updated. As a result, any transient 404 in CloudDataFusionPipelineStateSensor.poke() raises an unhandled HTTPError and fails the task attempt instead of returning False.

3. Deferrable Mode (CloudDataFusionStartPipelineOperator(deferrable=True) -> DataFusionAsyncHook in airflow/providers/google/cloud/hooks/datafusion.py):

PR #60688 narrowed except Exception: to except ValueError: in DataFusionAsyncHook._get_link() and DataFusionAsyncHook.get_pipeline_status(), whereas AioSession.get() raises aiohttp.ClientResponseError on HTTP 404 when the newly started pipeline run is not yet visible.
This PR fixes both def mode and sensor by catching correct exceptions:

The sensor calls DataFusionHook.get_pipeline_workflow(). That hook represents status 404 as requests.exceptions.HTTPError, so the sensor must catch HTTPError
Deferrable mode uses gcloud.aio.auth.AioSession, which raises aiohttp.ClientResponseError. It is caught where the async request is made, and only status 404 is retried
To proper distinguish 404 when the pipeline does not exist and when the pipeline is not visible yet, start_pipeline() must succeed and return a run_id. Only subsequent requests for that returned run_id treat 404 as transient.
If the pipeline itself does not exist, start_pipeline() returns 404 before polling begins, and that error still fails immediately.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[VladaZakharova]

@shahar1 @potiuk Hi there! Can i kindly ask you to review this PR?