ephraimbuddy commented on code in PR #68008:
URL: https://github.com/apache/airflow/pull/68008#discussion_r3594815338
##########
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:
This is still wrong and can overwrite already completed task. The heartbeat
scan does not lock the TI. Between that scan and ti.handle_failure, the worker
can commit the task as successful. handle_failure() refreshes the TI but does
not revalidate its state, so it overwrites that terminal SUCCESS with FAILED
after already enqueueing a failure callback.
--
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]
