Failures in Celery worker can set task in indefininte queued state

[frsann]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

An error in the Celery worker regularly leads to a situation where the scheduler "finishes" a task in a queued state, causing it to be queued indefinitely. Clearing the task fixes the situation. The situation show in the logs like this:

Apr 19 03:02:06.586 
scheduler_job.py:�[0m550} INFO�[0m - Sending TaskInstanceKey(dag_id='weather_forecasts_pipeline', task_id='extract_weatherbitio_forecast_daily', run_id='scheduled__2023-04-18T23:00:00+00:00', try_number=1, map_index=-1) to executor with priority 4 and queue high�[0m

Apr 19 03:02:06.586
executor.py:�[0m95} INFO�[0m - Adding to queue: ['airflow', 'tasks', 'run', 'weather_forecasts_pipeline', 'extract_weatherbitio_forecast_daily', 'scheduled__2023-04-18T23:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/weather_forecasts_pipeline.py']�[0m

Apr 19 03:03:33.517
[2023-04-19 00:03:32,972: WARNING/ForkPoolWorker-4] Dag 'weather_forecasts_pipeline' not found in path /usr/local/airflow/dags/weather_forecasts_pipeline.py; trying path /usr/local/airflow/dags/weather_forecasts_pipeline.py

Apr 19 03:04:34.550
[2023-04-19 00:04:33,611: ERROR/ForkPoolWorker-4] [21299cda-c9ed-4340-b778-a684943f09c4] Failed to execute task Dag 'weather_forecasts_pipeline' could not be found; either it does not exist or it failed to parse..

Apr 19 03:04:34.550
airflow.exceptions.AirflowException: Dag 'weather_forecasts_pipeline' could not be found; either it does not exist or it failed to parse.

Apr 19 03:04:35.681
uler_job.py:�[0m645} INFO�[0m - TaskInstance Finished: dag_id=weather_forecasts_pipeline, task_id=extract_weatherbitio_forecast_daily, run_id=scheduled__2023-04-18T23:00:00+00:00, map_index=-1, run_start_date=None, run_end_date=None, run_duration=None, state=queued, executor_state=failed, try_number=1, max_tries=2, job_id=None, pool=default_pool, queue=high, priority_weight=4, operator=ExtractWeatherbitIoData, queued_dttm=2023-04-19 00:02:05.820271+00:00, queued_by_job_id=19365049, pid=None�[0m

The fact that the dag file is not found is not the main point here, but rather the fact that the scheduler sets the task as finished with state=queued, executor_state=failed, and is not able to recover from this by retrying.

What you think should happen instead

I would expect that the task is set to "failed" state, and eventually retrying the task.

How to reproduce

In our case we are able to reproduce it by causing a AirflowExpection as the task is staring, by for example making the dag file unavailable.

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

Airflow version: 2.5.0

Anything else

No response

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.

[hussein-awala]

The fact that the dag file is not found is not the main point here

If you have a problem in dag parsing during the task execution (file not found, an exception while reading something from external system,...), the task is immediately marked as failed even if it didn't reach its max retries, because Airflow doesn't know if it's a temporary problem or you deleted the dag file and you want to stop the execution.

If the task state in the db is queued, and the parsing problem is temporary, #30375 may solve your problem, it will be available in 2.6.0, you can test it when it released.

But until then, I recommend to check what is the problem of the dag parsing, because it should be stable to ensure the stability of your Airflow server.

I close the issue, we can reopen it if needed.

[frsann]

We know the cause for the parsing errors and it has since been fixed, but the fact that the executor and the scheduler do not agree about the state of the task is a completely different issue. Also, how is "queued" a valid end-state for a task? We for example have alerting on high numbers of failed tasks, but this alert did not trigger due to the state being unexpected. Furthermore, as the DAG was in a running state it prevented future runs (unlike a failed DAG would have).

But thanks for the link to a potential fix. It taught me about the stalled_task_timeout configuration, which seems like a good fallback if this issue persist. Looking forward to test the fix in 2.6.0 👍

[potiuk]

We know the cause for the parsing errors and it has since been fixed, but the fact that the executor and the scheduler do not agree about the state of the task is a completely different issue.

FYI. It's a property of Airflow being distributed system and workers having to execute Python code. There is a DAG versioning Airflow Improvement proposal (it is in a draft form and has not been approved yet) https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-36+DAG+Versioning - and anyone in the community (your company ? ) is free to volunteer and implement it.

One of the properties of DAG versioning should be that the version of DAG Python file (and all releated imports) used to parse the structure of the DAG should be somewhat frozen for the whole execution time (including backfill execution of those if you choose so). That SHOUDL solve the problem, as long as you do not do a dynamic DAG generation with using external sources.

If you are using external sources (database, Pulling configuration etc.) while the DAG is parsed, there is absolutely nothing Airflow can do, and this is something that is considered as a very bad practice that you should never, ever do https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#top-level-python-code, especially that there are ways of doing it that are recommended https://airflow.apache.org/docs/apache-airflow/stable/howto/dynamic-dag-generation.html

So yeah, this might be another issue, but if you are violating best practices, this is your issue, not Airflow's.

Also, how is "queued" a valid end-state for a task

It should not be. And in Airflow 2.6.0 (that is going to be released in a few days most likely) this has been already fixed ( @hussein-awala mentioned it above) and we even have a nice blog explaining it published yesterday:

https://medium.com/apache-airflow/unsticking-airflow-stuck-queued-tasks-are-no-more-in-2-6-0-6f40a1a22835

BTW. Airflow 2.6.0rc1 will be out in a few days, and subscribing to devlist and testing it and reporting any problems is a great way you can help with getting things improved.