udsy19 opened a new pull request, #73291:
URL: https://github.com/apache/airflow/pull/73291
`MetastoreBackend.get_connection()` and `get_variable()` are decorated with
`@provide_session`. Called with no session — the shape every public caller
uses, including `BaseHook.get_connection(conn_id)` and `Variable.get(key)` from
a `dag_run` listener hook
(`on_dag_run_running`/`on_dag_run_success`/`on_dag_run_failed`) — the decorator
resolves the session to `settings.Session()`, a thread-local **scoped** session.
The scheduler's own `_do_scheduling` loop runs `dag_run` listener hooks
in-process (`DagRun.notify_dagrun_state_changed`) while holding that exact same
scoped session under `prohibit_commit`, a guard that raises on any unexpected
commit to protect its HA locking. The `@provide_session` wrapper commits on
exit, so a listener that reads a Connection or Variable trips the guard with
`RuntimeError("UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!")`. That error is
swallowed per-backend inside
`Connection.get_connection_from_secrets`/`Variable.get_variable_from_secrets`,
so it first surfaces as a spurious `AirflowNotFoundException` for a
Connection/Variable that genuinely exists — and then corrupts the scheduler's
session for the rest of that scheduling pass, observed cascading into
`DetachedInstanceError` on unrelated `DagRun` objects processed in the same
batch.
This is the root cause of #39646: a listener plugin calling
`BaseHook.get_connection()`/`Variable.get()` from
`on_dag_run_running`/`on_dag_run_success`/`on_dag_run_failed` gets
`AirflowNotFoundException` for connections/variables that are defined, while
the identical call from `on_task_instance_*` (which runs in a separate worker
process, not the scheduler) works fine.
Impact: crash-on-valid-input
WHO reaches this / entry point: any dag_run listener plugin registered the
documented way (a `hookimpl` implementing
`on_dag_run_running`/`on_dag_run_success`/`on_dag_run_failed`, exactly as shown
in the Listeners how-to doc and in #39646's own repro) that calls the standard
`BaseHook.get_connection(conn_id)` or `Variable.get(key)` — no session
argument, which is the only shape a plugin author has, since the hookspec
passes no session. Triggered by: the scheduler reaching
`SchedulerJobRunner._do_scheduling` ->
`_schedule_all_dag_runs`/`_start_queued_dagruns` ->
`DagRun.notify_dagrun_state_changed` for any DagRun transitioning to
running/success/failed while such a listener is registered -- an entirely
ordinary scheduling pass, valid input by construction (a real,
already-committed Connection/Variable). What they observe: first a wrong
`AirflowNotFoundException` for a Connection/Variable that exists, then (per the
DetachedInstanceError repro below) the scheduler's own session bre
aks for the rest of that scheduling pass.
## Fix
Give `get_connection()`/`get_variable()` a genuinely independent, non-scoped
session (`create_session(scoped=False)`) when no session is passed, instead of
the scoped one `@provide_session`'s default would resolve to. A listener's read
no longer aliases into — and commits — the scheduler's locked session.
`@provide_session` itself can't be reused for this (its `NEW_SESSION` path is
exactly the scoped session that's the problem), so the session handling is
inlined; the `check-new-session-in-provide-session` prek hook's None-typed
exemption (`session: Session | None = None`) covers the new shape.
This is a narrower, session-less-caller-focused companion to #71968/#72121,
which added an optional session-*reuse* parameter for internal scheduler call
sites that already have a session in scope
(`Variable.get_variable_from_secrets`). Neither of those covers
`BaseHook.get_connection()`/`Variable.get()` called with no session at all,
which is the public, session-agnostic shape every dag_run listener (and every
non-scheduler caller) actually uses — #71968's own body names
`Connection.get_connection_from_secrets` as exactly this left-open follow-up.
## Testing
Two new tests in `TestMetastoreBackendSessionSafety`
(`tests/unit/always/test_secrets_metastore.py`), mirroring the
`prohibit_commit`-under-guard pattern from #67980's
`task_instance_mutation_hook` regression tests:
- `test_get_connection_survives_prohibit_commit_without_explicit_session`
- `test_get_variable_survives_prohibit_commit_without_explicit_session`
Both reproduce the exact scheduler shape (`with create_session() as session:
with prohibit_commit(session):
MetastoreBackend().get_connection(...)`/`get_variable(...)`, no `session=`
argument) and assert the lookup succeeds.
Negative control: reverting only `metastore.py` to `main` and running the
class (`pytest
tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety`):
```
tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety::test_get_connection_survives_prohibit_commit_without_explicit_session
FAILED
tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety::test_get_variable_survives_prohibit_commit_without_explicit_session
FAILED
E RuntimeError: UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!
2 failed, 5 passed in 5.04s
```
With the fix, all 7 tests in the class pass (the 5 pre-existing tests plus
the 2 new ones), `7 passed`. Also re-verified the original issue's exact
scenario end to end against a real `SchedulerJobRunner._do_scheduling()` run
with a registered `on_dag_run_success` listener calling
`BaseHook.get_connection()`/`Variable.get()` on a real, committed
Connection/Variable: fails with `DetachedInstanceError` on pristine `main`,
passes with this fix.
closes: #39646
Signed-off-by: Udaya Tejas <[email protected]>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Sonnet 5)
Generated-by: Claude Code (Sonnet 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]
