L022937 opened a new issue, #73311:
URL: https://github.com/apache/airflow/issues/73311
### Apache Airflow version
3.3.1
### What happened
The scheduler crashes with `DetachedInstanceError` when scheduling DAG runs
that trigger mapped-task expansion. The crash occurs at `taskmap.py:271`:
```python
ti.context_carrier = new_task_run_carrier(dr.context_carrier)
```
where `dr = unmapped_ti.dag_run` (line 252) is a lazy-loaded DagRun that has
been detached from the SQLAlchemy session.
### The `list()` fix in 3.3.0 (#68634) is insufficient
Commit `1b6e47afa5` added `list()` around
`get_running_dag_runs_to_examine()` at `_do_scheduling:1899`, which
materializes the outer DagRun objects. However, the crash occurs on a
**different** DagRun object loaded via `unmapped_ti.dag_run` inside
`expand_mapped_task`, which is not covered by the `list()` materialization.
### Made FATAL by the error handler
The `except Exception:` handler at `scheduler_job_runner.py:2793` accesses
`run.run_id` on the same detached DagRun, causing a **second**
`DetachedInstanceError` that propagates past the catch block and kills the
scheduler process. Even if the original error were recoverable, the handler
turns it into a fatal crash.
### Root Cause
At `taskmap.py:252`, `dr` is obtained via `unmapped_ti.dag_run` — a lazy
relationship load. By the time `dr.context_carrier` is accessed at line 271,
the session has been expired/committed (via `guard.commit()` or nested session
operations inside `_schedule_dag_run` → `update_state` → `_get_ready_tis`),
detaching the DagRun from the session.
### What you think should happen instead
The scheduler should not crash. Proposed fixes (any of):
1. **Eagerly load `context_carrier`** when fetching the DagRun at
`taskmap.py:252-258` — add `.options(load_only(DagRun.context_carrier))` or use
`session.get()` to ensure the attribute is loaded before the session can expire
it
2. **Re-bind the DagRun** to the session before accessing `context_carrier`:
`dr = session.merge(dr)` at line 270
3. **Harden the error handler** at `scheduler_job_runner.py:2793` to not
access detached attributes — use `getattr(run, 'run_id', '<unknown>')` or catch
`DetachedInstanceError` in the handler
### How to reproduce
1. Deploy Airflow 3.3.1 with CeleryExecutor + KubernetesExecutor in
multi-team mode
2. Have multiple running DAG runs (>10) with at least some DAGs containing
mapped tasks (`.expand()`)
3. The scheduler will intermittently crash when it processes a running DAG
run that triggers `_expand_mapped_task_if_needed`
The crash is more likely with:
- Higher `max_dagruns_per_loop_to_schedule` (more concurrent DagRun
processing)
- DAGs with mapped tasks in the same scheduling batch as non-mapped DAGs
- After scheduler restarts (orphan task adoption can expire session objects)
### Full traceback
```
Traceback (most recent call last):
File "scheduler_job_runner.py", line 2788, in _schedule_all_dag_runs
callback = self._schedule_dag_run(run, session=session)
File "scheduler_job_runner.py", line 2903, in _schedule_dag_run
schedulable_tis, callback_to_run = dag_run.update_state(session=session,
execute_callbacks=False)
File "dagrun.py", line 1197, in update_state
info = self.task_instance_scheduling_decisions(session=session)
File "dagrun.py", line 1371, in task_instance_scheduling_decisions
schedulable_tis, changed_tis, expansion_happened =
self._get_ready_tis(...)
File "dagrun.py", line 1593, in _get_ready_tis
new_tis = _expand_mapped_task_if_needed(schedulable)
File "dagrun.py", line 1567, in _expand_mapped_task_if_needed
expanded_tis, _ = TaskMap.expand_mapped_task(ti.task, self.run_id,
session=session)
File "taskmap.py", line 271, in expand_mapped_task
ti.context_carrier = new_task_run_carrier(dr.context_carrier)
File "sqlalchemy/orm/attributes.py", line 569, in __get__
return self.impl.get(state, dict_)
File "sqlalchemy/orm/loading.py", line 1607, in load_scalar_attributes
raise orm_exc.DetachedInstanceError(...)
sqlalchemy.orm.exc.DetachedInstanceError: Instance <DagRun> is not bound to
a Session
During handling of the above exception, another exception occurred:
File "scheduler_job_runner.py", line 2793, in _schedule_all_dag_runs
self.log.exception("Error scheduling DAG run %s of %s", run.run_id,
run.dag_id)
^^ ALSO DETACHED
sqlalchemy.orm.exc.DetachedInstanceError: Instance <DagRun> is not bound to
a Session
```
### Operating System
Amazon Linux 2 (EKS)
### Versions of Apache Airflow Providers
apache-airflow-providers-celery==3.23.1
apache-airflow-providers-cncf-kubernetes==10.21.0
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
- EKS with 16 tenants (multi-team mode)
- CeleryExecutor + KubernetesExecutor per team
- PostgreSQL 16 on RDS Multi-AZ
- ~20 concurrent running DAG runs
- Scheduler restarts every ~10-20 minutes due to this crash
### Anything else
This bug existed in 3.2.1 as well, where the streaming `ScalarResult` from
`get_running_dag_runs_to_examine()` made it worse. The `list()` fix in 3.3.0
(#68634) reduced the frequency but did not eliminate the crash because the
detachment occurs on a *different* DagRun object inside `expand_mapped_task`,
not the one from the outer iterator.
### Are you willing to submit PR?
No
### 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]
