hkc-8010 commented on code in PR #68008:
URL: https://github.com/apache/airflow/pull/68008#discussion_r3605540528
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3620,6 +3652,27 @@ def _purge_task_instances_without_heartbeats(
request,
)
self.executor.send_callback(request)
+
+ # This purge path leaves the executor's own "task finished but TI
still looked queued"
+ # handling in process_executor_events unreachable for this TI once
handle_failure() below
+ # moves it out of RUNNING, so the email notification has to be
sent from here directly.
+ if task is not None and task.email and (task.email_on_failure or
task.email_on_retry):
+ self.executor.send_callback(
+ EmailRequest(
+ filepath=ti.dag_model.relative_fileloc or "",
+ bundle_name=bundle_name,
+ bundle_version=bundle_version,
+ version_data=version_data,
+ ti=ti,
+ msg=msg,
+ email_type=(
+ "retry" if task_callback_type ==
TaskInstanceState.UP_FOR_RETRY else "failure"
+ ),
+ context_from_server=context_from_server,
+ )
+ )
+
+ ti.handle_failure(error=msg, session=session)
Review Comment:
Good catch. Fixed in 3c9eef3d2f9 by closing the window on both sides.
The scan now locks the rows it selects (`with_row_locks(query, of=TI,
session=session, skip_locked=True)` in
`_find_task_instances_without_heartbeats`), so a worker can't commit a terminal
state on the same TI between the scan and the `handle_failure()` that runs
later in the same transaction. That matches how `process_executor_events` and
the other TI-mutating paths in this file already lock.
As defense in depth (and the piece that's actually unit-testable), the purge
loop now revalidates each row's committed state before doing anything and skips
it if it's no longer RUNNING/RESTARTING, so a task that completed concurrently
isn't clobbered and gets no failure callback. Added
`test_heartbeat_timeout_skips_ti_completed_concurrently` (worker wins the race,
assert no callback + state stays SUCCESS) and
`test_heartbeat_timeout_scan_locks_rows`. Validated on both sqlite and postgres.
--
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]
