zhengxiang378928908-code opened a new pull request, #74658:
URL: https://github.com/apache/airflow/pull/74658
closes: #59166
### Problem
When the metadata DB connection drops while
`update_dag_parsing_results_in_db` records import
errors (or Dag warnings), the Dag processor exits, and under a restart
policy it goes into a crash loop.
Reproduced on `main` with MySQL 8.0 by killing the session's connection
(`KILL CONNECTION_ID()` from another connection) inside
`_update_import_errors`:
1. `_update_import_errors` fails with `(2013, 'Lost connection to MySQL
server during query')`.
2. The `except` logs "Error logging import errors!" and swallows it, leaving
the session in a
failed transaction.
3. The final `session.flush()` has nothing to flush, so it doesn't raise and
the function returns
normally. The `try/except` around `persist_parsing_result` (#65313) is
therefore never triggered.
4. `handle_parsing_result` returns, `@provide_session` calls
`session.commit()`
(`utils/session.py:43`), which raises `PendingRollbackError: Can't
reconnect until invalid
transaction is rolled back`.
5. Nothing above `_collect_results` catches it, so the processor exits.
The same thing happens on any database for a non-connection error during the
flush (e.g. an integrity
error), because the session is again left needing a rollback.
### Fix
- `collection.py`: wrap the two best-effort writes (import errors, Dag
warnings) in
`session.begin_nested()`. A failure there now rolls back only its
savepoint, so the Dags and
serialized Dags written earlier in the same call are kept and the session
stays usable. If the
savepoint can't recover the session (connection invalidated, or session no
longer active), the
error is re-raised instead of being swallowed.
- `manager.py`: when persisting fails, `handle_parsing_result` now calls
`session.rollback()` before
continuing, so the teardown commit no longer raises and other files keep
being processed (the
existing throttling path is unchanged).
I did not use the `session.rollback()` inside
`update_dag_parsing_results_in_db` proposed in the
issue: that would also discard the Dags already written in the same
transaction, which turns a crash
into silently dropped Dag updates.
### Tests
- `test_handle_parsing_result_survives_failed_flush_in_own_session`:
`handle_parsing_result` without
a caller-supplied session, with a failing flush inside
`_update_import_errors`. Fails on `main`
with `PendingRollbackError` from `create_session`, passes with this change.
- `test_failed_import_error_write_keeps_dags_and_session_usable`: the Dag
row survives a failed
import-error write and the session is still active. Fails on `main`.
- `test_failed_import_error_write_reraises_when_session_unusable`: an
invalidated connection
propagates. It passes on `main` too (the error surfaces later there); it
guards against the
savepoint swallowing it, which I confirmed happens without the re-raise
check.
`tests/unit/dag_processing/test_manager.py` and `test_collection.py` pass on
both SQLite and
MySQL 8.0 (mysqlclient). I also ran the `KILL CONNECTION_ID()` reproduction
above against MySQL
locally, before and after the change; that script is not included in the PR.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Claude Opus 5.5)
Generated-by: Claude Code (Claude Opus 5.5) 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]
