Issue with task_queued_timeout Causing Silent Task Failures in Airflow 2.10.5

[KarthikeyanDevendhiran]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.10.5

What happened?

After upgrading to Airflow 2.10.5, we received the following deprecation warning:

/home/airflow/.local/lib/python3.10/site-packages/airflow/cli/commands/scheduler_command.py:41 DeprecationWarning: The '[kubernetes_executor] worker_pods_pending_timeout' config option is deprecated. Please update your config to use '[scheduler] task_queued_timeout' instead.

• We updated our configuration by setting task_queued_timeout = 300 under the [scheduler] section. A few days later, we noticed that several tasks in our production DAGs were failing after remaining in the queued state for 5 minutes. However, no failure alerts were triggered, even though the DAGs are configured to retry 3 times and send alerts via Slack.

• We assumed that since the tasks never started execution, the alerting mechanism might not have been triggered. However, our understanding is that task_queued_timeout should mark the task as failed and trigger alerts.

• To mitigate the issue, we increased the timeout to 900 seconds (15 minutes). This worked for about three weeks, but we are now seeing the same issue again—tasks are failing after being queued for over 15 minutes, without any alerts. This is happening despite having 100-node capacity available in our Databricks instance pool.

Either the task_queued_timeout is not integrated with the alerting mechanism or the tasks are not being scheduled properly despite available resources, possibly due to a scheduling or executor issue.

What you think should happen instead?

• Tasks that fail due to task_queued_timeout should trigger failure alerts, just like any other task failure.
• Tasks should not remain in the queued state for extended periods if sufficient executor capacity is available.

How to reproduce

• Upgrade to Airflow 2.10.5.
• Set task_queued_timeout = 300 in airflow.cfg under [scheduler].
• Deploy a DAG with tasks that may experience scheduling delays.
• Observe tasks failing after 5 minutes in the queued state without triggering alerts.
• Increase timeout to 900 seconds and observe recurrence under similar conditions.

Operating System

linux/amd64

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

We maintain our helm charts in a git repo and deploy them in Kubernetes using ArgoCD. All our tasks use Databricks compute to run the tasks.

Anything else?

• Does not appear to be a callback code error, as it works for other tasks/DAGs.
• Logs for the failed task show:
  *** Could not read served logs: Invalid URL 'http://:8793/log/dag_id=...': No host supplied

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[karenbraganz]

I was able to reproduce this issue as well. I am seeing logs for on_failure_callback in 2.10.0 but not in 2.10.5 when tasks get stuck in the queued state and fail.

[2025-06-02T21:01:54.409+0000] {processor.py:791} INFO - Executed failure callback for <TaskInstance: dag_1.task_1 scheduled__2025-06-01T00:00:00+00:00 [failed]> in state failed

I believe this is happening due to the changes introduced in PR #43520. With this PR, tasks don't immediately fail when the task_queued_timeout is met. Instead, they can requeue and re-attempt to run up to two more times (even without retries set).

In this PR, once the task instance fails to run after the two requeue attempts, its state will be set to failed directly with the TaskInstance set_state method.

Previously, tasks were failed by calling the cleanup_stuck_queued_tasks method. This would eventually result in the change_state function being called, which would add the task instance's failed state to the executor's event buffer . I think this step might have been important to ensure that the scheduler could process the task failure as an executor event and call ti.handle_failure() for the callback to be executed.

The solution might be to call the executor's fail method, which would call the change_state method similar to what was done before instead of using set_state to fail the task.

@dimberman or @RNHTTR is my understanding correct? I'd be happy to submit a PR to fix it.

[KarthikeyanDevendhiran]

@karenbraganz , I appreciate your insights. However, I'm puzzled as to why tasks remain in a pending state despite having sufficient instances available in my Databricks instance pool. While I experienced this issue occasionally with the previous Airflow version, it has become more frequent since upgrading to 2.10.5. For instance, out of 20 DAGs, I encounter this problem with 2-4 DAGs every three days, where the pods get stuck in a queued state and ultimately fail. Is this a common occurrence with the new update, or should I be concerned about this behavior?

[karenbraganz]

@KarthikeyanDevendhiran Are you saying that the tasks remain stuck in the queued state despite workers being available? This should not be a common occurrence with the upgrade to 2.10.5 especially since PR #43520 was created to reduce the incidence of this issue. With this PR, tasks are requeued up to two times by default with the same try number once the task_queued_timeout is reached. I can look into this if you provide the following information:

• Scheduler logs showing the lifecycle of a task instance that got stuck in queued and failed. Please filter the scheduler logs by adding the DAG ID, task ID, and run ID to your search query.
• Data from the log table in the Airflow metadata DB for the same task instance that got stuck in queued and failed. You may use this SQL query to get this information:

SELECT *
FROM log
WHERE dag_id = '<fill-in-dag-id>'
  AND task_id = '<fill-in-task-id>'
  AND run_id = '<fill-in-run-id>'
ORDER BY dttm ASC;

[KarthikeyanDevendhiran]

Hi @karenbraganz , Apologies for the delayed response. I’ve shared the scheduler logs and Airflow metadata logs for the two tasks that were marked as failed but did not trigger any Slack alerts. Please find the requested logs below for your review.

Scheduler logs

`kubectl logs airflow-scheduler-65cfcf89c9-w8xjc -n airflow | grep "TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00'"
Defaulted container "scheduler" out of: scheduler, git-sync, scheduler-log-groomer, wait-for-airflow-migrations (init), git-sync-init (init)
[2025-06-09T10:00:00.476+0000] {scheduler_job_runner.py:692} INFO - Sending TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) to KubernetesExecutor with priority 1 and queue default
[2025-06-09T10:00:00.485+0000] {kubernetes_executor.py:352} INFO - Add task TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) with command ['airflow', 'tasks', 'run', 'dag_name', 'task_name', 'scheduled__2025-06-09T08:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/folder_name.py']
[2025-06-09T10:00:05.201+0000] {scheduler_job_runner.py:776} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1)
[2025-06-09T10:00:12.165+0000] {kubernetes_executor_utils.py:425} INFO - Creating kubernetes pod for job is TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1), with pod name dag_name-task_name-8vpdausn, annotations: <omitted>
[2025-06-09T10:15:48.070+0000] {scheduler_job_runner.py:692} INFO - Sending TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) to KubernetesExecutor with priority 1 and queue default
[2025-06-09T10:15:48.075+0000] {kubernetes_executor.py:352} INFO - Add task TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) with command ['airflow', 'tasks', 'run', 'dag_name', 'task_name', 'scheduled__2025-06-09T08:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/folder_name.py']
[2025-06-09T10:15:48.714+0000] {scheduler_job_runner.py:776} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1)
[2025-06-09T10:15:49.006+0000] {kubernetes_executor_utils.py:425} INFO - Creating kubernetes pod for job is TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1), with pod name dag_name-task_name-rkrqu4hr, annotations: <omitted>
[2025-06-09T10:31:50.179+0000] {scheduler_job_runner.py:692} INFO - Sending TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) to KubernetesExecutor with priority 1 and queue default
[2025-06-09T10:31:50.184+0000] {kubernetes_executor.py:352} INFO - Add task TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1) with command ['airflow', 'tasks', 'run', 'dag_name', 'task_name', 'scheduled__2025-06-09T08:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/folder_name.py']
[2025-06-09T10:31:50.255+0000] {kubernetes_executor_utils.py:425} INFO - Creating kubernetes pod for job is TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1), with pod name dag_name-task_name-rx89exlt, annotations: <omitted>
[2025-06-09T10:31:51.614+0000] {scheduler_job_runner.py:776} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='dag_name', task_id='task_name', run_id='scheduled__2025-06-09T08:00:00+00:00', try_number=1, map_index=-1)`

Airflow Metadata logs

TASK 1

`airflow=> SELECT * FROM log WHERE dag_id = 'dag_name' AND task_id = 'task_name' AND run_id = 'scheduled__2025-06-09T08:00:00+00:00' LIMIT 10;
   id    |             dttm              |       dag_id       |         task_id         | map_index |             event
       | execution_date | owner |                                               extra
      | owner_display_name |                run_id                | try_number
---------+-------------------------------+--------------------+-------------------------+-----------+-------------------------
-------+----------------+-------+---------------------------------------------------------------------------------------------
------+--------------------+--------------------------------------+------------
1802783 | 2025-06-09 10:15:47.409503+00 | dag_name | task_name |        -1 | stuck in queued reschedu
le     |                |       | Task was in queued state for longer than 900.0 seconds; task state will be set back to sched
uled. |                    | scheduled__2025-06-09T08:00:00+00:00 |          1
1802824 | 2025-06-09 10:31:49.016565+00 | dag_name | task_name |        -1 | stuck in queued reschedu
le     |                |       | Task was in queued state for longer than 900.0 seconds; task state will be set back to sched
uled. |                    | scheduled__2025-06-09T08:00:00+00:00 |          1
1802844 | 2025-06-09 10:47:50.486896+00 | dag_name | task_name |        -1 | stuck in queued tries ex
ceeded |                |       | Task was requeued more than 2 times and will be failed.
      |                    | scheduled__2025-06-09T08:00:00+00:00 |          1`

TASK 2

`id    |             dttm              |        dag_id        |      task_id      | map_index |             event
   | execution_date | owner |                                               extra
  | owner_display_name |                run_id                | try_number
---------+-------------------------------+----------------------+-------------------+-----------+-----------------------------
---+----------------+-------+-------------------------------------------------------------------------------------------------
--+--------------------+--------------------------------------+------------
1699360 | 2025-05-02 14:48:53.464319+00 | dag_name | task_name |         -1 | stuck in queued reschedule
   |                |       | Task was in queued state for longer than 900.0 seconds; task state will be set back to scheduled
. |                    | scheduled__2025-04-30T14:00:00+00:00 |          1
1699381 | 2025-05-02 15:04:55.373482+00 | dag_name | task_name |         -1 | stuck in queued reschedule
   |                |       | Task was in queued state for longer than 900.0 seconds; task state will be set back to scheduled
. |                    | scheduled__2025-04-30T14:00:00+00:00 |          1
1699400 | 2025-05-02 15:21:02.064989+00 | dag_name | task_name |         -1 | stuck in queued tries exceed
ed |                |       | Task was requeued more than 2 times and will be failed.
  |                    | scheduled__2025-04-30T14:00:00
`

Let me know if you need any additional details. Thanks!!!

[karenbraganz]

@KarthikeyanDevendhiran there are two separate issues being discussed here. We should discuss them separately to avoid confusion.

• Tasks that fail after being stuck in queued do not trigger a failure callback. Since this is reproducible, I will leave this issue open until we are able to fix it.
• Increased occurrence of stuck in queued tasks in 2.10.5. Since this is not reproducible and we are still gathering information on this, it would be more appropriate to create a discussion for this. I will create a separate discussion for us to continue looking into it.

[karenbraganz]

@KarthikeyanDevendhiran I have created discussion #51597 to discuss the increased incidence of stuck in queued tasks. Please provide the requested information on that discussion thread.

As for the issue of stuck in queued tasks not triggering failure callbacks- the solution I proposed in this comment works for Airflow 2 but not Airflow 3.

I implemented and tested my proposed changes on Astro Runtime 3.0-2 (Airflow 3.0.1) but saw the below error when the callback was executed:

[2025-06-10T18:14:13.466+0000] {manager.py:147} INFO - DAG bundles loaded: main
--- Supervised process Last chance exception handler ---
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/supervisor.py", line 369, in _fork_main
    target()
  File "/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 90, in _parse_file_entrypoint
    result = _parse_file(msg, log)
             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 106, in _parse_file
    _execute_callbacks(bag, msg.callback_requests, log)
  File "/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 146, in _execute_callbacks
    raise NotImplementedError(
NotImplementedError: Haven't coded Task callback yet - https://github.com/apache/airflow/issues/44354!

I implemented and tested the same changes in Astro Runtime 13.0.0 (Airflow 2.11.0), and saw that the failure callback executed successfully. Prior to my changes, the callback was not executed when tasks got stuck in queued and failed.

I believe this discrepancy between the Runtime versions might be due to breaking changes introduced in Airflow 3. Therefore, we might have to implement different solutions for Airflow 2 vs 3. I will keep looking into this.

[karenbraganz]

I believe the NotImplementedError is happening in Airflow 3 because we have not actually implemented a callback workflow for tasks that get stuck in queued or fail due to heartbeat timeouts in the Airflow 3 code. I have asked Ash for confirmation on this on the OSS Slack.

I will work on a fix for Airflow 2 only, and once I receive more information from Ash, we can decide on next steps to fix this in Airflow 3.