edumuellerFSL opened a new issue, #71172:
URL: https://github.com/apache/airflow/issues/71172
### 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"` is marked failed by the scheduler
with a state mismatch, where the executor event being processed belongs to the
previous poke of the same try.
To be clear about 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 report is about `BaseSensorOperator(mode="reschedule")`, where the
task exits `up_for_reschedule` and is re-queued for the next poke. Same shape
of race, different path, and the existing branch does not cover it.
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:02.838 worker Executing workload in Celery
00:04:03.003 api-server Task started (previous_state=queued)
00:04:07.339 api-server Task instance state updated
new_state=up_for_reschedule
00:04:07.466 worker Task finished exit_code=0
final_state=up_for_reschedule
(celery task returns normally, result is SUCCESS)
00:05:07.338 (db) task_reschedule.reschedule_date for the next poke
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 itself exited cleanly (`exit_code=0`,
`final_state=up_for_reschedule`). Its executor success was not drained until
89s later, against a 60s `poke_interval`, by which point the scheduler had
already put the same TI back to `scheduled` for the next poke 3.5s earlier.
The scheduler line immediately before the error, which shows both sides:
```
TaskInstance Finished: dag_id=..., task_id=..., run_id=..., ti_id=...,
run_start_date=... 00:04:02.935, run_end_date=... 00:04:07.339,
run_duration=4.403,
state=scheduled, executor=CeleryExecutor(parallelism=512),
executor_state=success,
try_number=1, max_tries=0, operator=ExternalTaskSensor,
queued_dttm=... 00:04:02.121, scheduled_dttm=... 00:05:33.399,
queued_by_job_id=J
```
Notice `scheduled_dttm` (00:05:33.399) is earlier than the moment the event
was processed (00:05:36.901), and `try_number` is the same on both sides. And
the resulting error:
```
Executor CeleryExecutor(parallelism=512) reported that the task instance
<TaskInstance: <dag_id>.<sensor_task_id> scheduled__... [scheduled]
ti_id=...>
finished with state success, but the task instance's state attribute is
scheduled.
```
**Why the current `ti_requeued` branch does not cover this**
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, `ti.try_number == buffer_key.try_number`
holds and `ti_queued` is True.
All three conditions in `ti_requeued` are False for a reschedule-mode sensor:
- `ti.queued_by_job_id != job_id` is False when the same scheduler that
queued poke N also re-queues poke N+1. With 8 schedulers it is usually a
different job id, which is why this fires intermittently rather than on every
poke.
- `executor.has_task(ti)` is False. The reschedule date has passed and the
TI is back in `scheduled`, but it has not been handed to the executor yet.
- the resume-after-defer condition is False, because `ti.next_method` is
`None` on a reschedule exit. It is only set when resuming from a trigger.
So it falls through to `if ti_queued and not ti_requeued` and the task is
failed.
Checked against `main` (e69c188) and 3.3.0. On `main` the resume-after-defer
condition was widened to cover `QUEUED` as well as `SCHEDULED`, but it is still
gated on `ti.next_method is not None`. Nothing in `ti_requeued` references
`TaskReschedule` or `up_for_reschedule`.
Conditions that have to line up:
1. a sensor with `mode="reschedule"`
2. executor event drain latency greater than `poke_interval`
3. the same scheduler processes both the re-queue and the stale event
4. the TI is in `scheduled` (or `queued`) when the stale success is drained
**Reproducer (probabilistic)**
Same caveat as #66374 and #67287: this is a race and is not
deterministically reproducible.
1. Airflow 3.2.x or later, any executor whose event buffer can lag.
2. A sensor with `mode="reschedule"` and a short `poke_interval` (60s here)
whose condition stays unmet for many pokes.
3. Put the scheduler under enough load that draining the executor event
buffer takes longer than `poke_interval`.
4. Watch the audit log for the `[scheduled]` vs `success` pair on the sensor
task.
We see it on CeleryExecutor, but nothing in the mechanism is
Celery-specific: it only needs the executor event for poke N to arrive after
the scheduler has re-queued poke N+1. #66374 hit the analogous defer-path race
on CeleryExecutor and #67287 hit it on LocalExecutor, so the family already
spans both.
**Differences and linkages with current tickets**
- #23824 / #23846 (CLOSED): the original 2.x race between triggerer and
scheduler, fixed by "Do not fail requeued TIs". Same underlying shape, defer
path.
- #66374 (CLOSED): the 3.x scheduled-state variant of the defer path. Fixed
by #66431, backported to `v3-2-test` by #67089.
- #67287 (CLOSED): the queued-state variant of the same defer path, which
#66431 did not cover. Fixed by widening the condition to `ti.state in
(SCHEDULED, QUEUED)`.
- This report: the reschedule-mode sensor path. Both existing fixes are
gated on `ti.next_method is not None`, which is never true for a reschedule
exit, so no existing condition applies. The state pair and the failure mode are
otherwise identical to #66374.
### 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 now is, rather than as
an externally killed task.
The distinguishing signal is the presence of a `TaskReschedule` row for the
TI, which is what tells the scheduler this task exited `up_for_reschedule` at
least once rather than being killed:
```python
ti_requeued = (
ti.queued_by_job_id != job_id
or executor.has_task(ti)
or (
# Resume-after-defer (existing)
ti.state in (TaskInstanceState.SCHEDULED, TaskInstanceState.QUEUED)
and state == TaskInstanceState.SUCCESS
and ti.next_method is not None
)
or (
# Stale reschedule-exit: the poke exited up_for_reschedule and the
next poke was
# already re-queued before we drained the executor success for the
same try_number.
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
)
)
```
Two notes on that:
The `TaskReschedule` lookup is what keeps the condition narrow. Simply
dropping the `next_method is not None` requirement would also swallow genuine
external kills of `scheduled` TIs, which is the case this branch exists to
catch.
`TaskReschedule` is not currently referenced in `scheduler_job_runner.py`,
and doing a lookup per event would add a query to the hot path. It should be
resolved once for the TIs in the buffer, alongside the existing bulk TI fetch
in `_process_executor_events`, rather than per event. Happy to shape it
whichever way 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, across roughly 13 different
DAG authors. Every one is a sensor in `mode="reschedule"`; no other operator
type produced this state pair.
Frequency tracks scheduler and metadata database latency rather than
anything in the DAGs. On the worst day, when the metadata DB was CPU-saturated
by an unrelated long-running transaction, 5 of the 25 landed within that day.
The affected tasks run with `max_tries=0`, so a single stale event ends the
DAG run rather than costing a retry.
### 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)
--
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]
