miquelmartorell opened a new issue, #72716:
URL: https://github.com/apache/airflow/issues/72716
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.1 (checked against `main`, unchanged)
### What happened and how to reproduce it?
Another variant of the same family, on the defer path: the resumed attempt
is already
`RUNNING` when the stale executor event is drained.
A deferrable sensor defers. The worker process exits cleanly on `defer()`,
and the
executor reports `SUCCESS` for that try. The trigger fires, the scheduler
requeues the
task instance, and the worker starts the resumed attempt — the TI reaches
`RUNNING` and
clears `next_method`. Only *then* does the scheduler drain the stale
executor success
event from the defer-exit. In `process_executor_events`, `ti_queued` is True
(`RUNNING`
is in the tuple) and `ti_requeued` is False, so the TI is logged as `state
mismatch` and
sent to `handle_failure`.
Timeline of one occurrence (UTC, single scheduler, `LocalExecutor`, several
hundred
deferrable sensors in one DAG run):
04:01:22.031 TI -> deferred (worker exits; executor will
report success)
04:01:23.377 trigger fired, success
04:01:25.648 TI -> running (resumed attempt starts,
next_method cleared)
04:01:28.516 state mismatch (stale defer-exit success
drained here)
04:01:28.520 TI -> failed
04:01:33.967 resumed worker heartbeat -> 409, process killed
Executor LocalExecutor(parallelism=32) reported that the task instance
<TaskInstance: <dag_id>.<sensor_task_id> ... [running] ti_id=...>
finished with state success, but the task instance's state attribute is
running.
Why the current branch does not catch it:
- `ti.queued_by_job_id != job_id` is False — the same scheduler handled both.
- `executor.has_task(ti)` is False — both attempts share the same
`TaskInstanceKey`
(`try_number` does not increment on resume), and the defer-exit event has
already
removed that key from the executor's bookkeeping.
- The resume-after-defer condition is False on two counts: it requires
`ti.state in (SCHEDULED, QUEUED)` and `ti.next_method is not None`, and by
the time the
resumed attempt is running the state is `RUNNING` and `next_method` has
been cleared.
So it falls through and the task is failed. The in-code comment above the
guard says the
`RUNNING` case "is handled by the scheduler detecting task instances without
heartbeats",
which does not apply here: the resumed attempt is alive and heartbeating
normally — the
problem is the stale event, not a missing heartbeat.
With `retries: 0` on the affected tasks there is no second attempt, so a
task whose
deferred work completed successfully ends as `failed` and takes the DAG run
with it.
Not deterministically reproducible: anything that makes the scheduler slower
to drain the
event buffer widens the window. It shows up with
- one DAG with several hundred deferrable sensors and `retries: 0`,
- an executor sharing the host with the scheduler,
- a DAG file whose import is slow enough that many workers are starting at
once,
competing for the same CPU as the scheduler loop.
The deterministic fingerprint is a `state mismatch` row in `log` immediately
followed by
`failed`, for a task whose trigger event was `success`.
**Differences and linkages with current tickets**
- #23824 / #23846 (CLOSED): the original 2.x version of this race, defer
path.
- #66374 (CLOSED): the scheduled-state variant, defer path. Fixed by #66431,
backported
by #67089.
- #67287 (CLOSED): the queued-state variant of the same defer path, fixed by
#68741.
- #71172 (OPEN): the reschedule-mode sensor path. Same shape of race,
different path; the
fix proposed there is gated on a `TaskReschedule` row, which never exists
here.
- This one: the defer path where the resumed attempt has already reached
`RUNNING`. Both
merged fixes are gated on `ti.state in (SCHEDULED, QUEUED)` **and**
`next_method is not None`, so neither applies.
### What you think should happen instead?
A stale executor success from a defer-exit should never fail a task instance
that the
trigger has already resumed, regardless of whether the resumed attempt is
`SCHEDULED`,
`QUEUED` or already `RUNNING`.
Widening the state tuple is not enough on its own, because `next_method` is
already
cleared once the resumed attempt runs, so neither field identifies the case
any more.
Two directions that would:
1. Make the stale event stop matching: bump the attempt identifier on resume
so the
defer-exit event no longer matches `buffer_key` for the current attempt.
2. Compare timestamps: ignore an executor success whose TI is `RUNNING` and
whose
current attempt started after the event was produced.
### Operating System
Ubuntu 22.04 (Linux 5.15)
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
Not provider-specific. Reproduced with astronomer-cosmos 1.15.1 deferrable
sensors, but
the mechanism is executor/scheduler-side and provider-independent.
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
Custom image built on the official Airflow base image; no changes to
scheduler,
executor or Task SDK code.
### Anything else?
Twice in three days on the same DAG, on a different sensor each time. Our
workaround is
to stop deferring those sensors, which removes the defer-exit event entirely.
### Are you willing to submit PR?
- [ ] 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]
