udsy19 opened a new pull request, #73358: URL: https://github.com/apache/airflow/pull/73358
## Summary closes: #46224 When a deferrable operator's trigger ends its task directly with a `TaskSuccessEvent`/`TaskFailedEvent` (no resume on a worker), the `BaseTaskEndEvent` handler in `airflow-core/src/airflow/models/trigger.py` already routes a `FAILED` event through retry-eligibility and sends the matching `on_failure_callback`/`on_retry_callback` (#69821 fixed that half). It never sends an `EmailRequest`, though, so `email_on_failure` and `email_on_retry` are silently never honored for a task that ends this way, even though its callbacks fire correctly and nothing in the UI or logs indicates anything was skipped. Every other place a task instance ends without going back to a worker — the scheduler's executor-event handling and its externally-killed-task handling, both in `airflow-core/src/airflow/jobs/scheduler_job_runner.py` — already sends an `EmailRequest` alongside the callback request; the trigger-driven path is the only one of the three that did not. Who reaches this: any Dag author who sets `email`/`email_on_failure`/ `email_on_retry` (all default `True`) on a task built with a deferrable operator whose trigger can end the task itself via `TaskSuccessEvent`/ `TaskFailedEvent` — the documented ["Exiting a deferred task from a trigger"](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#exiting-deferred-task-from-triggers) pattern. Triggered whenever such a task fails or is retried; the failure notification email documented for every operator simply never arrives. Impact: silent-wrong-result ## Fix Send an `EmailRequest` from the `BaseTaskEndEvent` handler for a `FAILED`/ `UP_FOR_RETRY` outcome, using the task's `email`/`email_on_failure`/ `email_on_retry` configuration exactly as the two scheduler sites already do, and the same `DatabaseCallbackSink` the handler already uses to send the `TaskCallbackRequest` two lines above (the executor isn't available at this call site, but `DatabaseCallbackSink` is the same sink `executor. send_callback` resolves to by default). ## Negative control Added `test_submit_event_task_end_failed_sends_email` (parametrized over the retry-eligible and retries-exhausted cases) to the pre-existing `test_trigger.py`, alongside `test_submit_event_task_end_failed_respects_retries` which #69821 added for the callback half of this same mechanism. It submits a `TaskFailedEvent` to a deferred task instance with `email` configured and asserts exactly one `EmailRequest` (not just the `TaskCallbackRequest`) was sent, with the retry-vs-failure-correct `email_type`. Test fails without the fix, passes with it: ``` without the fix (trigger.py reverted to upstream/main): 2 failed AssertionError: expected exactly one EmailRequest to be sent, got calls=[...TaskCallbackRequest...] assert 0 == 1 with the fix: 2 passed ``` Full `test_trigger.py` suite still green after the fix: 46 passed, no regression. ## Was generative AI tooling used to co-author this PR? - [x] Yes (please specify the tool below) Claude Code Generated-by: Claude Code following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) Signed-off-by: Udaya Tejas <[email protected]> -- 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]
