apache / apache/airflow

Scheduler bulk-fails all scheduled tasks when serialized DAG is transiently missing

Open
#62,050 3 comments 0 reactions 0 assignees View on GitHub
affected_version:3.2 area:core area:scheduler area:serialization kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Apache Airflow version

Other Airflow 3 version (please specify below)

### If "Other Airflow 3 version" selected, which one?

3.0.6, still on main

### What happened?

My instance has been hitting intermittent task failures on MWAA (Airflow 3.0.6, ~150 DAGs, 4 schedulers). Tasks failed in bulk with no obvious cause but succeeded on manual retry. I noticed this on `scheduler_job_runner.py`:

When the scheduler can't find a DAG in the `serialized_dag` table, it does this:

```python
session.execute(
update(TI)
.where(TI.dag_id == dag_id, TI.state == TaskInstanceState.SCHEDULED)
.values(state=TaskInstanceState.FAILED)
.execution_options(synchronize_session="fetch")
)
```

It sets every SCHEDULED task instance for that DAG to FAILED.

With PR #58259 and #56422, it probably happens less often but the bulk-failure issue has never been addressed

### What you think should happen instead?

The scheduler could skip scheduling that DAG for the current iteration and try again next time, instead of immediately failing everything.

I thought of logging a warning instead of error, tracking a counter per DAG, and only failing tasks after several consecutive misses to distinguish transient gaps from genuinely missing DAGs. What do you think about this solution?

PR #55126 tried something similar for stale DAGs (skipping and continuing)

### How to reproduce

```python
def test_missing_serialized_dag_bulk_fails(self, dag_maker, session):
dag_id = "SchedulerJobTest.test_missing_serialized_dag_bulk_fails"

with dag_maker(dag_id=dag_id, session=session):
EmptyOperator(task_id="task_a")
EmptyOperator(task_id="task_b")

scheduler_job = Job()
self.job_runner = SchedulerJobRunner(job=scheduler_job)

# Simulate serialized DAG being transiently missing
self.job_runner.scheduler_dag_bag = mock.MagicMock()
self.job_runner.scheduler_dag_bag.get_dag_for_run.return_value = None

dr = dag_maker.create_dagrun(state=DagRunState.RUNNING, run_type=DagRunType.SCHEDULED)
for ti in dr.task_instances:
ti.state = State.SCHEDULED
session.merge(ti)
session.flush()

res = self.job_runner._executable_task_instances_to_queued(max_tis=32, session=session)
session.flush()

assert len(res) == 0
tis = dr.get_task_instances(session=session) # Both tasks are FAILED instead of SCHEDULED
for ti in tis:
print(f"{ti.task_id}: {ti.state}")
```

### Operating System

AWS MWAA

### Versions of Apache Airflow Providers

n/a

### Deployment

Amazon (AWS) MWAA

### Deployment details

- MWAA environment running 3.0.6 (latest available on MWAA as of Feb 2026)
- mw1.2xlarge instance class
- 4 schedulers
- 5-20 workers (auto-scaling)
- ~150 DAGs
- min_file_process_interval=300

### Anything else?

would appreciate guidance on the preferred approach (retry counter or just ignoring)

### 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)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.