seanmuth opened a new issue, #52264: URL: https://github.com/apache/airflow/issues/52264
### Apache Airflow version 2.11.0 ### If "Other Airflow 2 version" selected, which one? 2.9, 2.10 ### What happened? During handling of an `on_failure_callback` in the dag processor, the callback failed due to a deadlock on the TI record in question. Because we only call `session.flush()` and `session.commit()` [L768-9](https://github.com/apache/airflow/blob/d9ed7b94d033dc61ee1c05aafb9f1c1f7b82cfb2/airflow/dag_processing/processor.py#L768-769] _outside_ of the `for request in callback_requests:` loop, the session remains in an unusable state. This particular DAG had _hundreds_ of TIs fail, due to a worker OOM issue, which subsequently caused every callback in the `callback_requests` list to fail to run, because the session was never trashed and a new one opened. ### What you think should happen instead? A new session should be created for _each_ callback in the loop, or short of that, if we fall to the exception block we should trash the session there and create a new one. ### How to reproduce 1. induce multiple on_failure_callbacks 2. during callback processing, another component must hold a lock on a TI while the callback is also trying to execute for that TI ### Operating System debian ### Versions of Apache Airflow Providers _No response_ ### Deployment Astronomer ### Deployment details Runtime 11.18.0 / Airflow 2.9.3+astro.11 ### Anything else? _No response_ ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md) -- 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]
