apache / apache/airflow

Race condition between scheduler processing events and sensor reschedule

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

Description

### Under which category would you file this issue?

Airflow Core

### Apache Airflow version

3.2.2 (checked against 3.3.0 and `main`, unchanged)

### What happened and how to reproduce it?

A sensor running with `mode="reschedule"` gets marked failed by the scheduler, because the executor result for one poke is processed after the next poke has already been re-queued.

A note on the word "reschedule", since it is overloaded here: this is not the defer path. #66431 is titled "ignore stale executor success after defer reschedule", where reschedule means the scheduler re-queued a deferred TI. This is about `BaseSensorOperator(mode="reschedule")`. Same shape of race, different path.

Timeline of one occurrence (UTC, `poke_interval=60`, CeleryExecutor):

```
00:04:02.121 scheduler TI queued (try_number=1, queued_by_job_id=J)
00:04:03.003 worker Task started
00:04:07.339 worker up_for_reschedule, next poke due 00:05:07
00:04:07.466 worker Task finished exit_code=0 final_state=up_for_reschedule
(celery task returns normally, result is SUCCESS)
00:05:33.399 scheduler TI put back to scheduled for the next poke
00:05:36.901 scheduler executor success from the 00:04:07 poke is drained
00:05:36.914 scheduler ERROR state mismatch, task marked FAILED
```

The poke exited cleanly. Its executor success took 89s to drain, against a 60s poke interval, so by the time the scheduler looked at it the same TI was already back in `scheduled` for the next poke.

```
Executor CeleryExecutor(parallelism=512) reported that the task instance
. ... [scheduled] ti_id=...>
finished with state success, but the task instance's state attribute is scheduled.
```

Why the current branch does not catch it:

https://github.com/apache/airflow/blob/e69c1881b32c36abb827bae3717eaf46424427bd/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1529-L1539

In reschedule mode `try_number` does not increment between pokes, so poke N and poke N+1 share an executor key and `ti_queued` is True. Then `ti.queued_by_job_id != job_id` is False because the same scheduler handled both (with 8 schedulers it is usually a different one, which is why this is intermittent rather than constant), `executor.has_task(ti)` is False because the TI is back in `scheduled` and has not been handed to the executor yet, and the resume-after-defer condition is False because `ti.next_method` is `None` on a reschedule exit. So it falls through and the task is failed.

Not deterministically reproducible: same DAG, same sensor, most pokes are fine and occasionally one races. Anything that makes the scheduler slower to drain the event buffer widens the window. We see it on CeleryExecutor, but nothing in the mechanism is Celery-specific, and the defer variant of this family was reported on both Celery (#66374) and Local (#67287).

Checked against 3.3.0 and `main`. On `main` the resume-after-defer condition now covers `QUEUED` as well as `SCHEDULED`, but it is still gated on `ti.next_method is not None`, and nothing in `ti_requeued` references `TaskReschedule`.

**Differences and linkages with current tickets**

- #23824 / #23846 (CLOSED): the original 2.x version of this race, defer path.
- #66374 (CLOSED): the 3.x scheduled-state variant, defer path. Fixed by #66431, backported by #67089.
- #67287 (CLOSED): the queued-state variant of the same defer path.
- This one: the reschedule-mode sensor path. The existing fixes are all gated on `next_method is not None`, which is never true for a reschedule exit, so none of them apply.

### What you think should happen instead?

A stale executor success belonging to a completed reschedule poke should be treated as a requeue, the same way the defer-exit case is now, rather than as an externally killed task.

The signal that distinguishes it is a `TaskReschedule` row for the TI, which is what says the task exited `up_for_reschedule` rather than being killed. Roughly:

```python
or (
ti.state in (TaskInstanceState.SCHEDULED, TaskInstanceState.QUEUED)
and state == TaskInstanceState.SUCCESS
and ti.next_method is None
and ti.id in ti_ids_with_reschedule
)
```

Without the `TaskReschedule` part this would also swallow genuine external kills, which is what the branch is there to catch. `TaskReschedule` is not referenced in `scheduler_job_runner.py` today and a per-event lookup would sit on the hot path, so it probably wants resolving once alongside the existing bulk TI fetch. Happy to shape it however maintainers prefer.

### Operating System

Debian GNU/Linux 12 (bookworm), official `apache/airflow:slim-3.2.2-python3.12` image

### Deployment

Official Apache Airflow Helm Chart

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.20.0
apache-airflow-task-sdk==1.2.2

### Official Helm Chart version

1.21.0

### Kubernetes Version

v1.34.2

### Helm Chart configuration

CeleryExecutor, 8 scheduler replicas, `core.parallelism=512`, `celery.worker_concurrency=16`, `celery.task_acks_late=false`. No other scheduler-related overrides.

### Docker Image customizations

Official slim image plus in-house provider packages installed with `uv`. No changes to `airflow-core`; the scheduler is stock.

### Anything else?

25 occurrences over 7 weeks in one deployment, spread across about 13 different DAG authors, and every one is a sensor in `mode="reschedule"`. It tracks scheduler and metadata database latency rather than anything in the DAGs: on our worst day, with the database CPU-saturated by an unrelated long-running query, 5 of the 25 landed within that day.

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

Research direction

Start in airflow-core/src/airflow/jobs/scheduler_job_runner.py at the executor-result handling around the linked lines, then trace how ti_requeued and the existing bulk TI fetch are populated. Check how TaskReschedule identifies a completed sensor reschedule, and verify that a stale SUCCESS for such a poke is treated as a requeue while genuine external kills still fail the task.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend, data-engineering, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.