wolfier opened a new issue, #56196: URL: https://github.com/apache/airflow/issues/56196
### Apache Airflow version 3.1.0 ### If "Other Airflow 2 version" selected, which one? _No response_ ### What happened? `on_failure_callback` of a task with remaining retry attempts executed when the worker executing the task terminated abruptly. This is not expected because the task has retry attempts. ### What you think should happen instead? When a task with remaining retries fails, the `on_retry_callback` should be executed and **NOT** `on_failure_callback`. ### How to reproduce 1. Execute a long running task 2. Terminate the worker 3. Allow the scheduler to identify the task instance was killed externally 4. Examine dag processor logs for execution of the task's`on_failure_callback` ### Operating System Debian ### Versions of Apache Airflow Providers _No response_ ### Deployment Astronomer ### Deployment details _No response_ ### Anything else? When a task with either `on_failure_callback` or `on_retry_callback` is killed externally, a [TaskCallbackRequest](https://github.com/apache/airflow/blob/3.1.0/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L930C31-L942) is sent. The request does not have `task_callback_type` set and defaults to `None`. The callback request is considered a [failure callback](https://github.com/apache/airflow/blob/3.1.0/airflow-core/src/airflow/callbacks/callback_requests.py#L69-L72) when `task_callback_type` is `None`. Since the `task_callback_type` is `None`, the task's `on_failure_callback` is queued for execution when the dag processor processes the callback request. ```python if request.task_callback_type is TaskInstanceState.UP_FOR_RETRY: callbacks = task.on_retry_callback else: callbacks = task.on_failure_callback ``` The `TaskCallbackRequest` should set `task_callback_type` according to whether the task is eligible to retry. If it is, the task should set `task_callback_type` to `TaskInstanceState.UP_FOR_RETRY` otherwise `TaskInstanceState.FAILED` is an appropriate mode. ```python request = TaskCallbackRequest( filepath=ti.dag_model.relative_fileloc, bundle_name=ti.dag_version.bundle_name, bundle_version=ti.dag_version.bundle_version, ti=ti, msg=msg, context_from_server=TIRunContext( dag_run=DRDataModel.model_validate(ti.dag_run, from_attributes=True), max_tries=ti.max_tries, variables=[], connections=[], xcom_keys_to_clear=[], ), task_callback_type=( TaskInstanceState.UP_FOR_RETRY if ti.is_eligible_to_retry() else TaskInstanceState.FAILED ), ) ``` ### 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) -- 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]
