dsuhinin commented on code in PR #69336:
URL: https://github.com/apache/airflow/pull/69336#discussion_r3806387799
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1516,7 +1516,17 @@ def process_executor_events(
)
)
- if ti_queued and not ti_requeued:
+ # A running task that's still sending heartbeats is alive -- a
worker is running it right now.
+ # This event is probably from a duplicate that already lost and
died, so don't fail the live
+ # run. If the task really did die, heartbeat detection will fail
it once the heartbeat stops.
+ heartbeat_timeout = conf.getint("scheduler",
"task_instance_heartbeat_timeout")
+ ti_alive = (
+ ti.state == TaskInstanceState.RUNNING
Review Comment:
Thanks for digging into this! I don't think it's a no-op though(but I could
be mistaken) — the two conditions look at different things.
The `state` we unpack on L1488 comes out of `event_buffer.pop(buffer_key)`.
That buffer is filled by the **executor** — every
`success()`/`fail()`/`queued()`/`running_state()` call writes `(state, info)`
into it. So `state` here is "what the executor reported", not the TI's DB
state. `buffer_key` is just the
lookup key into that dict.
`ti_alive`, on the other hand, checks `ti.state`, which we loaded from the
database in the `session.scalars(...)` query above.
So passing the L1490 `continue` only tells us the **executor** didn't report
QUEUED/RUNNING — it says nothing about what's in the DB. By the time we reach
`ti_alive` we're handling a **terminal** executor event (SUCCESS/FAILED), and
at that moment `ti.state` can still be `RUNNING`. That's could be exactly the
duplicate-dispatch case this guards against: a live worker is actually
running the task and heartbeating, while a stale terminal event lands from the
duplicate that already died. Without the check we'd fail the live run out from
under the worker.
I can add a short comment making the `state` (executor) vs `ti.state` (DB)
distinction explicit if that'd help future readers.
--
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]
