hkc-8010 opened a new pull request, #76908: URL: https://github.com/apache/airflow/pull/76908
When a task has been requeued `num_stuck_in_queued_retries` times and is still stuck in queued, `_maybe_requeue_stuck_ti` sets it to `failed` with `ti.set_state(FAILED)`. That skips `retries` entirely. On Airflow 3 the DAG processor callback only runs user callbacks and never touches state, so the task stays `failed` for good. The callback request also had no `task_callback_type`, so `on_failure_callback` fired even when retries were left. This changes the tries-exceeded branch to do what the heartbeat-timeout purge already does: - load the task and set `ti.task`, so `is_eligible_to_retry()`, `retry_delay` and `fail_fast` work - send the `TaskCallbackRequest` with `UP_FOR_RETRY` or `FAILED`, built from the failed attempt before it is archived - send an `EmailRequest` from here, because the executor event for this TI no longer reaches the path in `process_executor_events` that sends one - call `ti.handle_failure()` instead of `set_state(FAILED)`, then `executor.fail()` with the failed attempt's key The stale executor event is dropped as before: on retry the old row is archived, and with no retries left the TI is already `failed`, not `queued`. ### How this showed up On 2.11.2 the same `set_state(FAILED)` produces a different symptom. There the DAG processor's callback does call `handle_failure()`, so the task does retry a few seconds later. But in that gap the scheduler sees a failed upstream and marks `all_success` downstream tasks `upstream_failed`, and nothing re-evaluates them after the retry succeeds. The run then finishes `success` with those tasks skipped. This was reproduced with the docker-compose setup on 2.11.2 (worker concurrency 1, `task_queued_timeout=60`, one task holding the only slot). It did not reproduce on 2.8.4, where the timeout went through the executor-event failure path and never left a `failed` window. On main there is no gap, because nothing ever retries the task. ### Behaviour changes - `handle_failure()` writes the usual `failed` Log row, increments the failure stats and fires `on_task_instance_failed`. Listeners such as OpenLineage now see a FAILED event for these tasks, as they would for any other failure. - With retries left, the old executor event is discarded with the existing "no longer current" warning. ### Tests - `test_stuck_in_queued_tries_exceeded_honours_retries` (0 and 1 retries): final state, callback type, email type, failed attempt id and try number in both requests, the executor key passed to `fail()`, and with a retry pending the downstream task is not `upstream_failed`. - `test_stuck_in_queued_tries_exceeded_retries_without_callbacks`: retries still happen with no callbacks, and when the DAG can't be loaded. - Two existing tests now expect the extra `failed` Log row. Reverting the source change fails 6 tests. Dropping only the callback type, only the email, or putting `set_state(FAILED)` back each fails at least 2. `test_scheduler_job.py` passes on Python 3.11 (609 passed). The stuck-in-queued tests also pass on Postgres 14. No newsfragment, since this is a bug fix with no interface change. Happy to add one if reviewers want it. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
