Dask executor doesn't get correct response if worker restarted during task running

[danxian-baiheng]

Apache Airflow Provider(s)

standard

Versions of Apache Airflow Providers

apache-airflow-providers-daskexecutor==1.1.1
apache-airflow-providers-fab==1.4.1
apache-airflow-providers-ftp==3.11.1
apache-airflow-providers-http==4.13.1
apache-airflow-providers-imap==3.7.0
apache-airflow-providers-smtp==1.8.0
apache-airflow-providers-sqlite==3.9.0
apache-airflow-providers-common-compat==1.2.1
apache-airflow-providers-common-io==1.4.2
apache-airflow-providers-common-sql==1.18.0

Apache Airflow version

2.10.2

Operating System

Debian GNU/Linux 12 (bookworm)

Deployment

Other Docker-based deployment

Deployment details

Kubernetes: v1.28
GoVersion:"go1.15.9"

What happened

./dask_executor.log-2025-01-12 08:30:47,368 - distributed.worker - INFO - Closing worker gracefully: tcp://{IP:port}. Reason: worker-lifetime-reached
./dask_executor.log-[2025-01-12T15:30:47.438+0000][271075][MainProcess] {{task_command.py:467}} INFO - Running <TaskInstance: {dag_id}.{task_id}.import manual__2025-01-12T15:28:42+00:00 [queued]> on host
./dask_executor.log-2025-01-12 08:30:47,497 - distributed.worker - INFO - Stopping worker at tcp://{IP:port}. Reason: worker-lifetime-reached
./dask_executor.log:2025-01-12 08:30:47,498 - distributed.worker.state_machine - WARNING - Async instruction for <Task cancelled name='execute("TaskInstanceKey(dag_id=\'...\', task_id=\'...\', run_id=\'manual__2025-01-12T15:28:42+00:00\', try_number=1, map_index=-1)")' coro=<Worker.execute() done, defined at /opt/nucolumnaradmin/.venv/lib/python3.11/site-packages/distributed/worker_state_machine.py:3610>> ended with CancelledError
./dask_executor.log-2025-01-12 08:30:47,499 - distributed.core - INFO - Connection to ... has been closed.
./dask_executor.log-2025-01-12 08:30:55,887 - distributed.nanny - INFO - Worker closed

Show as above logs, the dask worker triggered gracefully restart when a airflow task still running.
After that, the task's status keep in 'pending' and will no longer changed.
The running task's context is a shell cmd started with [airflow run ...], it will keep running and mark itself as success in airflow's metadata DB.

...
[2025-01-12T15:30:54.978+0000][271103][MainProcess] {{taskinstance.py:352}} INFO - Marking task as SUCCESS. dag_id=..., task_id=..., run_id=manual__2025-01-12T15:28:42+00:00, execution_date=20250112T152842, start_date=20250112T153048, end_date=20250112T153054
[2025-01-12T15:30:55.048+0000][271075][MainProcess] {{local_task_job_runner.py:266}} INFO - Task exited with return code 0

Their date are actually a same value with different time zone.

After that, this airflow task's status in airflow DB was success and will no longer be scheduled, at the mean time, its status in Dask executor(in airflow scheduler process) will keep in 'pending' until airflow scheduler restarted and will affect value of executor_slots_available, when the executor_slots_available reach zero, the airflow scheduler will stop working.

The debug dump from DaskExecutor, actually no task is running during that time:

SIGUSR2 received, printing debug
--------------------------------------------------------------------------------
[2025-01-13T03:18:54.192+0000][683642][MainProcess] {{scheduler_job_runner.py:278}} INFO - Debug dump for the executor DaskExecutor(parallelism=0)
[2025-01-13T03:18:54.192+0000][683642][MainProcess] {{base_executor.py:627}} INFO - executor.queued (0)

[2025-01-13T03:18:54.193+0000][683642][MainProcess] {{base_executor.py:632}} INFO - executor.running (176)

What you think should happen instead

Obviously the above status is incorrect and looks like a memory leak Bug.

How to reproduce

Maybe airflow should sync task status in Dask executor with metadata DB and cleanup succeed tasks.
Or just make sure Dask executor can receive the same status as the actually running cmd.

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

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.

[danxian-baiheng]

I haven't found the DaskExecutor tag in airflow-provider's list.
Does it be removed from airflow in the latest version?

[potiuk]

I haven't found the DaskExecutor tag in airflow-provider's list. Does it be removed from airflow in the latest version?

Yes. We no longer publish nor fix DasK Executor, it has been removed from maintance as result of this LAZY CONSENSUS https://lists.apache.org/thread/fxv44cqqljrrhll3fdpdgc9h9fz5ghcy and you will find the detailed discussion on why we decided to do it. But if someone would like to for the code of last dask executor, look at any problems there and release a new version - likely with a 3rd-party managed dask provider, they are entirely free to do it.

[potiuk]

Converting to a discussion as it's at most that.