Tasks stuck in queued state

[benrifkind]

Apache Airflow version

2.5.0

What happened

Tasks are getting stuck in the queued state

What you think should happen instead

Tasks should get scheduled and run

How to reproduce

I am using the CeleryExecutor and deploying Airflow on AWS's EKS.

I have 3 DAGs with 10 tasks. Each task is a simple KubernetesPodOperator which just exits when it starts. If I run the Airflow deploy with CELERY__WORKER_CONCURRENCY set to something high like 32, the celery worker will fail and the tasks that were queued up to run on it will enter into a bad state. Even once I set the concurrency lower (16), the tasks continue to not be scheduled. Note that if I set the worker concurrency to 16 on the initial deploy the tasks never get into a bad state and everything works fine.

Clearing the tasks does not even fix the issue. I get this log line in the scheduler

ERROR - could not queue task TaskInstanceKey(dag_id='batch_1', task_id='task_2', run_id='scheduled__2022-01-07T04:05:00+00:00', try_number=1, map_index=-1) (still running after 4 attempts)

To me it seems like the scheduler thinks the task is still running even though it is not.

Clearing the task and restarting the scheduler seems to do the trick.

Happy to give any more information that would be needed. Tasks getting stuck in queued also sometimes happens in my production environment which is the impetus for this investigation. I'm not sure if it is the same problem but I would like to figure out if this is a bug or just a misconfiguration on my end. Thanks for your help.

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.1.0
apache-airflow-providers-cncf-kubernetes==5.0.0

Deployment

Other 3rd-party Helm chart

Deployment details

I am using this helm chart to deploy - https://github.com/airflow-helm/charts/tree/main/charts/airflow (v8.6.1)

I know that chart is not supported by Apache Airflow but don't think it's related to the chart. Based on the logs and the solution it seems like an issue with Airflow/Celery.

Anything else

This problem can be replicated each time following the steps I detailed above. Not sure if the way the celery worker fails matters.

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[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[uranusjr]

Since it seems the issue can be consistently replicated, have you tried to reduce an localise the source of the issue? For example, does this happen with other operators, or specific to KubernetesPodOperator? Fewer DAGs and/or fewer tasks? On another machine (e.g. your local PC with LocalExecutor)?

[okayhooni]

In my case, there is no relation between task stuck issue and operator type. It can occur to any operator class(built-in operators, provider official operators, or our custom operators).

We use Airflow setup similar to @benrifkind (CeleryExecutor / deployed on EKS). But, we cannot replicate this task stuck issue consistently. (we got randomly 0~3 stuck task instances per day out of 8,000 TI per day.)

• more detail: Tasks stuck in queued state #28714 (reply in thread)

[benrifkind]

Hi @uranusjr. Thanks for your response.

I believe this is an issue with the CeleryExecutor so I have not tested it with an other Executors.

I checked and this doesn't seem to be an issue specifically with KubernetesPodOperator. I was able to replicate it with the BashOperator.

In terms of number of DAGs and tasks. I was able to replicate this with one DAG with many tasks. I think the issue occurs when a celery worker goes down unexpectedly while it is still responsible for running tasks. So with one celery worker, running one DAG with a lot of tasks and high concurrency creates the problem. Basically the celery worker dies or is killed and once it comes back up the scheduler thinks the tasks are being run so it can't rerun them on this restarted worker. Of course I'm not sure that is what is happening but it's my best guess. I am not sure why restarting the scheduler after clearing the tasks fixes the issue.

Thanks for your help.

Edit: More context and supporting information.

I just ran into this problem in my production deployment. And was able to fix it by clearing the tasks and restarting the scheduler. We restart our entire Airflow deployment by just terminating the instances that our EKS deployment runs on once a week. This terminates the Airflow celery workers without waiting for them to shut down gracefully. And so this problem along with the solution likely matches the problem I documented above.

[potiuk]

Something to be investigated then - looks like a problem in handling such exceptional case with celery. Thanks for narrowing it down.. In the meantime - are you sure you want to shitdiwn the deployment that abruptly?

[benrifkind]

Yup. That's a good point. This is not a good way to shut down Airflow. I will try to move off this approach to restarting the deployment. Thanks for the feedback.

[potiuk]

Converted to a discussion. Maybe we will get an issue from that, maybe we won't - if there are any thoughts about it from anyone that will be useful.

[okayhooni]

Although this looks like a stopgap measure, I made a PR related this CeleryExecutor issues. (calling clear_not_launched_queued_tasks() on interval, like KubernetesExecutor)

• Add handling logic on CeleryExecutor to reschedule task stuck in queued status #28943

(PR description, like below)
We have been using Airflow with Celery executor on EKS to schedule up to 8,000 task instances per day on production data pipeline, and suffered from task instances stuck in queued status.

On average, we found 1~3 task instances stuck in queued states per day, and those tasks did not conform execution_timeout or write any task logs, because they had not been launched by any celery worker!

I found logic for handling similar issues on KubernetesExecutor's method: clear_not_launched_queued_tasks()

• https://github.com/apache/airflow/blob/main/airflow/executors/kubernetes_executor.py#L495

So, I had made a custom DAG and task to perform similar operation(find tasks stuck in queued status, and re-schedule those by updating state field to scheduled with sqlalchemy query (like {TaskInstance.state: State.SCHEDULED})

this DAG works very well for the last few months, but on last Friday, we got very funny situation that task instance of this custom DAG was stuck in queued state. so, I guess this checking logic will be better if it is located on the executor code side with airflow built-in sched based scheduler(EventScheduler), like KubernetesExecutor.

If we know the root cause of this problem, there will be more elegant solution. but, I read some related issues linked above, and guess there is not just one reason to make stuck tasks, case by case. (Some parts of this issue is marked as fixed, but the issues remains consistently to latest release (v2.5)

• Tasks stuck in queued state #21225 (comment)