VladaZakharova opened a new pull request, #74237: URL: https://github.com/apache/airflow/pull/74237
Upstream refactoring PR https://github.com/apache/airflow/pull/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 https://github.com/apache/airflow/pull/60688 changed DataFusionHook.wait_for_pipeline_state() from except Exception: to except KeyError:. (was fixed in https://github.com/apache/airflow/pull/72406) 2. *Sensor Mode* (CloudDataFusionPipelineStateSensor.poke in airflow/providers/google/cloud/sensors/datafusion.py): Prior to PR https://github.com/apache/airflow/pull/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 https://github.com/apache/airflow/pull/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 https://github.com/apache/airflow/pull/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. <!-- SPDX-License-Identifier: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0 --> <!-- Thank you for contributing! Please provide above a brief description of the changes made in this pull request. Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/ Please make sure that your code changes are covered with tests. And in case of new features or big changes remember to adjust the documentation. For user-facing UI changes, please attach before/after screenshots (or a short screen recording) so reviewers can assess the visual impact. Feel free to ping (in general) for the review if you do not see reaction for a few days (72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications. In case of an existing issue, reference it using one of the following: * closes: #ISSUE * related: #ISSUE --> --- ##### Was generative AI tooling used to co-author this PR? <!-- If generative AI tooling has been used in the process of authoring this PR, please change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by". --> - [ ] Yes (please specify the tool below) <!-- Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) --> --- * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
