Abolfazlrwm opened a new pull request, #76629: URL: https://github.com/apache/airflow/pull/76629
<!-- SPDX-License-Identifier: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0 --> closes: #63374 related: #63692 (stale-closed draft whose approach this PR builds on, with credit to @Sathvik-Chowdary-Veerapaneni), #69991 ##### Why are the changes needed? Several users report (on 3.1.7 and 3.2.2) that a DAG-level `on_failure_callback` never fires, while the scheduler logs `callback is empty` for every DagRun. Root cause: `DagRun.update_state()` produces the `DagCallbackRequest` via `DagRun.produce_dag_callback()`, and building its `DagRunContext` lazy-loads `consumed_asset_events` inside a pydantic validator. That validator currently recovers only from `DetachedInstanceError`. When the DagRun is attached to a session whose transaction was invalidated by a prior database error, the lazy-load raises `PendingRollbackError`; a closed session raises `InvalidRequestError`. Both escape the narrow catch, so the exception aborts `update_state()` and the callback request is never produced — the scheduler then falls through to the `callback is empty` debug line. This is exactly the location @kaxil pointed to in the review of #69991 ("doesn't touch `DagRun.update_state()`, where the issue says the callback goes missing"). Note that @ieCecchetti could not reproduce the issue in clean environments, which is consistent with a session/timing-dependent race. This PR therefore hardens the callback-production path so transient ORM state problems can no longer silently drop a DAG-level callback, and makes the actually-sent case visible in logs. The failure modes are simulated directly in unit tests. ##### How are the changes implemented? - `airflow-core/src/airflow/callbacks/callback_requests.py` — `DagRunContext._sanitize_consumed_asset_events`: - catch `SQLAlchemyError` instead of `DetachedInstanceError`; - read the DagRun PK through `inspect(dag_run).identity` so the recovery itself cannot re-raise on expired state; - reload the DagRun with eagerly-loaded relationships on a fresh session and hand that freshly loaded instance to the validation instead of the broken one — a session invalidated by a failed flush has **every** attribute expired, so restoring only the relationship would still leave the conversion failing on the remaining expired columns (verified empirically); if the row is gone, degrade gracefully to an empty relationship list. - `airflow-core/src/airflow/jobs/scheduler_job_runner.py` — `_send_dag_callbacks_to_processor` now logs at debug level when a Dag callback **is** sent to the DAG processor, so the remaining `callback is empty` line can no longer be misread as proof that the callback was dropped. ##### Tests - `test_dagrun_context_consumed_asset_events_with_invalidated_transaction`: poisons a session with a failed flush (`IntegrityError`, then `PendingRollbackError` on the relationship lazy-load) and asserts the `DagRunContext` is still produced. - `test_dagrun_context_reload_with_deleted_dagrun`: the reload finds no row and the context is still produced with an empty relationship list. Executed locally: - `pytest airflow-core/tests/unit/callbacks/test_callback_requests.py` → 22 passed - `pytest airflow-core/tests/unit/models/test_dagrun.py` → 328 passed, 2 skipped (backend-specific), 1 xfailed - `pytest airflow-core/tests/unit/jobs/test_scheduler_job.py -k callback` → 50 passed - `prek run --files <changed files>` → 56 hooks passed; the breeze-dependent `check-tests-in-the-right-folders` hook was skipped via `SKIP_BREEZE_PREK_HOOKS=1` because breeze is not available in my environment (the test file location itself is unchanged from upstream) --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Z.ai GLM, an AI coding agent used for code research, implementation and test authoring, following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions). -- 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]
