[v2-11-test] Allow failure callbacks for stuck in queued TIs that fail by karenbraganz · Pull Request #53038 · apache/airflow

karenbraganz · 2025-07-08T18:52:08Z

closes: #51301
In issues #51301, it was reported that failure callbacks do not run for task instances that get stuck in queued and fail in Airflow 2.10.5. This is happening due to the changes introduced in PR #43520. In this PR, logic was introduced to requeue tasks that get stuck in queued (up to two times by default) before failing them.

Previously, the executor's fail method was called when the task needed to be failed after max requeue attempts. This was replaced by the task instance's set_state method in the PR ti.set_state(TaskInstanceState.FAILED, session=session). Without the executor's fail method being called, failure callbacks will not be executed for such task instances. Therefore, I changed the code to call the executor's fail method instead.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

kaxil · 2025-07-10T17:23:26Z

tests are failing:

FAILED tests/jobs/test_scheduler_job.py::TestSchedulerJob::test_handle_stuck_queued_tasks_multiple_attempts - AssertionError: assert equals failed
  [            [           
    'queued',    'failed', 
    'queued',    'failed', 
  ]            ]
FAILED tests/jobs/test_scheduler_job.py::TestSchedulerJob::test_handle_stuck_queued_tasks_reschedule_sensors - AssertionError: Expected 'fail' to not have been called. Called 2 times.
Calls: [call(TaskInstanceKey(dag_id='test_fail_stuck_queued_tasks_multiple_executors', task_id='op1', run_id='45e7135c-8e2c-4e07-8d35-4bea8c0232d6', try_number=0, map_index=-1)),
 call(TaskInstanceKey(dag_id='test_fail_stuck_queued_tasks_multiple_executors', task_id='op2', run_id='45e7135c-8e2c-4e07-8d35-4bea8c0232d6', try_number=0, map_index=-1))].

pytest introspection follows:

Args:
assert equals failed
  (                                ()                              
    TaskInstanceKey(dag_id='test_                                  
  fail_stuck_queued_tasks_multipl                                  
  e_executors', task_id='op2', ru                                  
  n_id='45e7135c-8e2c-4e07-8d35-4                                  
  bea8c0232d6', try_number=0, map                                  
  _index=-1),                                                      
  )
=========== 2 failed, 2035 passed, 959 skipped in 655.53s (0:10:55) ============

karenbraganz · 2025-07-14T00:47:42Z

@kaxil all tests are passing now.

airflow/jobs/scheduler_job_runner.py

potiuk

LGTM - providing that #53435 will also be merged :)

kaxil · 2025-07-22T19:14:35Z

LGTM - providing that #53435 will also be merged :)

Yeah, let's get that one merged first

eladkal · 2025-09-18T12:45:04Z

I see #53435 is merged

karenbraganz · 2025-09-20T00:46:00Z

@eladkal I had to make some changes there. Let me see if the changes are needed here as well.

potiuk · 2025-10-13T14:50:29Z

@eladkal I had to make some changes there. Let me see if the changes are needed here as well.

@karenbraganz -> did you check it ?

karenbraganz · 2025-10-14T19:19:02Z

@potiuk There are a few changes that I need to make. Working on the changes now.

karenbraganz · 2025-10-16T01:38:44Z

I still need to run some tests to ensure my changes work as expected.

karenbraganz · 2025-10-24T23:03:48Z

Just need to figure out why one of the checks is failing. I think it might have something to do with the rich-click package version being used.

Everything else is completed.

karenbraganz · 2025-10-27T20:31:58Z

I had to make changes to airflow/www/static/js/types/api-generated.ts because one of the CI hooks was failing. My PR is not related to the UI.

karenbraganz · 2025-10-27T20:34:11Z

Going to close and re-open the PR to re-trigger tests.

eladkal · 2026-02-09T09:54:47Z

@karenbraganz can you rebase and resolve conflicts?
cc @potiuk another candidate for 2.11.1

potiuk · 2026-02-11T14:36:03Z

I rebased your changes @karenbraganz -> @kaxil maybe you want to have a look before I merge it.

…ailure callbacks This commit addresses the handling of tasks that remain stuck in the queued state beyond the configured retry threshold. Previously, these tasks were marked as failed in the database but the executor was not properly notified, leading to inconsistent state between Airflow and the executor. Changes made: - Modified _maybe_requeue_stuck_ti() to accept executor parameter and call executor.fail() when tasks exceed requeue attempts, ensuring the executor is notified of task failures - Added logic to retrieve the DAG and task object to check for on_failure_callback - When a failure callback exists, create a TaskCallbackRequest and send it via executor to ensure failure callbacks are invoked for stuck queued tasks - Updated tests to verify that executor.fail() is called and callbacks are sent appropriately - Added test fixtures with mock_failure_callback to validate callback invocation - Changed pyproject.toml Python version requirement (unrelated) Why this matters: Tasks stuck in queued state can now be properly cleaned up at both the Airflow scheduler and executor levels, preventing resource leaks and ensuring failure callbacks are executed for proper error handling and alerting. Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

potiuk · 2026-02-13T15:09:07Z

Nicely green now... Merging.

…ailure callbacks (#53038) This commit addresses the handling of tasks that remain stuck in the queued state beyond the configured retry threshold. Previously, these tasks were marked as failed in the database but the executor was not properly notified, leading to inconsistent state between Airflow and the executor. Changes made: - Modified _maybe_requeue_stuck_ti() to accept executor parameter and call executor.fail() when tasks exceed requeue attempts, ensuring the executor is notified of task failures - Added logic to retrieve the DAG and task object to check for on_failure_callback - When a failure callback exists, create a TaskCallbackRequest and send it via executor to ensure failure callbacks are invoked for stuck queued tasks - Updated tests to verify that executor.fail() is called and callbacks are sent appropriately - Added test fixtures with mock_failure_callback to validate callback invocation - Changed pyproject.toml Python version requirement (unrelated) Why this matters: Tasks stuck in queued state can now be properly cleaned up at both the Airflow scheduler and executor levels, preventing resource leaks and ensuring failure callbacks are executed for proper error handling and alerting. Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

karenbraganz requested review from XD-DENG, ashb and kaxil as code owners July 8, 2025 18:52

karenbraganz self-assigned this Jul 8, 2025

eladkal added this to the Airflow 2.11.1 milestone Jul 11, 2025

karenbraganz mentioned this pull request Jul 16, 2025

Allow failure callbacks for stuck in queued TIs that fail #53435

Merged

kaxil reviewed Jul 18, 2025

View reviewed changes

airflow/jobs/scheduler_job_runner.py Show resolved Hide resolved

potiuk approved these changes Jul 21, 2025

View reviewed changes

karenbraganz changed the title ~~[v2-11-test] Call executor fail method for stuck in queued tasks~~ [v2-11-test] Allow failure callbacks for stuck in queued TIs that fail Aug 8, 2025

karenbraganz requested review from bbovenzi, jedcunningham, pierrejeambrun and ryanahamilton as code owners October 27, 2025 19:36

karenbraganz added the legacy ui Whether legacy UI change should be allowed in PR label Oct 27, 2025

karenbraganz closed this Oct 27, 2025

karenbraganz reopened this Oct 27, 2025

potiuk force-pushed the tqt-callbacks-2-11 branch from 1ffb9c0 to 875e831 Compare February 11, 2026 14:35

potiuk linked an issue Feb 11, 2026 that may be closed by this pull request

Issue with task_queued_timeout Causing Silent Task Failures in Airflow 2.10.5 #51301

Closed

2 tasks

potiuk approved these changes Feb 11, 2026

View reviewed changes

potiuk force-pushed the tqt-callbacks-2-11 branch 2 times, most recently from 1a78efc to 9c67a68 Compare February 13, 2026 11:28

potiuk force-pushed the tqt-callbacks-2-11 branch from 9c67a68 to abcedd7 Compare February 13, 2026 13:55

potiuk merged commit d5fbdd6 into apache:v2-11-test Feb 13, 2026
39 checks passed

potiuk mentioned this pull request Feb 17, 2026

Status of testing of Apache Airflow 2.11.1rc2 (and Fab Provider 1.5.4rc1) #62056

Closed

26 tasks

Conversation

karenbraganz commented Jul 8, 2025 • edited by eladkal Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kaxil commented Jul 10, 2025

Uh oh!

karenbraganz commented Jul 14, 2025

Uh oh!

Uh oh!

potiuk left a comment

Choose a reason for hiding this comment

Uh oh!

kaxil commented Jul 22, 2025

Uh oh!

eladkal commented Sep 18, 2025

Uh oh!

karenbraganz commented Sep 20, 2025

Uh oh!

potiuk commented Oct 13, 2025

Uh oh!

karenbraganz commented Oct 14, 2025

Uh oh!

karenbraganz commented Oct 16, 2025

Uh oh!

karenbraganz commented Oct 24, 2025

Uh oh!

karenbraganz commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

karenbraganz commented Oct 27, 2025

Uh oh!

eladkal commented Feb 9, 2026

Uh oh!

potiuk commented Feb 11, 2026

Uh oh!

potiuk commented Feb 13, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

karenbraganz commented Jul 8, 2025 •

edited by eladkal

Loading

karenbraganz commented Oct 27, 2025 •

edited

Loading