retries and callbacks are not honoured for ending deferred task from trigger
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
main (development)
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#exiting-deferred-task-from-triggers
On tasks with retry when a `TaskFailedEvent` is from trigger the task is marked as failed with the retries not executed for the failed tasks. Similar issue with callbacks and also email_on_failure, email_on_retry also not being honored. This happens in main and also 2.10.4 . PR related https://github.com/apache/airflow/pull/40084
### What you think should happen instead?
Retries, callbacks and emails on failure should be honored.
### How to reproduce
1. Create following triggers and dag with retries and callback.
2. Run the dag with `TaskFailedEvent` emitted from the trigger.
3. The dag is marked as failed without retries and callbacks executed.
4. Uncomment yielding `TriggerEvent` and run the dag again with new code.
5. Retry happens and the failure callback is also executed.
```python
# plugins/custom_trigger.py
from __future__ import annotations
import asyncio
import logging
from airflow.triggers.base import BaseTrigger, TriggerEvent, TaskSuccessEvent, TaskFailedEvent
from airflow.utils import timezone
class StateTrigger(BaseTrigger):
def __init__(self, state):
super().__init__()
self.state = state
def serialize(self):
return ("custom_trigger.StateTrigger", {"state": self.state})
async def run(self):
if self.state == "success":
yield TaskSuccessEvent()
else:
yield TaskFailedEvent()
# yield TriggerEvent(self.state)
```
```python
# dag_state_test.py
from __future__ import annotations
from datetime import datetime
from custom_trigger import StateTrigger
from airflow import DAG
from airflow.models.baseoperator import BaseOperator
class MultipleDeferTrigger(BaseOperator):
"""Multiple defer trigger."""
def __init__(self, state=None, *args, **kwargs):
self.state = state
super().__init__(*args, **kwargs)
def execute(self, context):
self.defer(
trigger=StateTrigger(self.state),
method_name="execute_complete",
)
def execute_complete(self, context, event=None):
raise Exception(event)
with DAG(
dag_id="state_defer",
start_date=datetime(2021, 1, 1),
catchup=False,
schedule=None,
default_args = {
"on_success_callback": lambda context: open("/tmp/on_success_callback", "w+").write(str(datetime.now())),
"on_failure_callback": lambda context: open("/tmp/on_failure_callback", "w+").write(str(datetime.now())),
"retries": 1,
"retry_delay": 5.0
}
) as dag:
success = MultipleDeferTrigger(task_id="success", state="success", retry_delay=5.0)
failed = MultipleDeferTrigger(task_id="failed", state="failed", retry_delay=5.0)
success
failed
```
### Operating System
Ubuntu 20.04.3 LTS
### Versions of Apache Airflow Providers
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### Anything else?
_No response_
### 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)
Contributor guide
Research direction
Start with the reproduction in plugins/custom_trigger.py and dag_state_test.py, focusing on BaseOperator.defer(), execute_complete(), and TaskFailedEvent. Reproduce the failed-trigger path, then trace deferred-task exit handling; done means retries, failure/retry callbacks, and failure/retry emails are honored as in the TriggerEvent path.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100