You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "danxian-baiheng (via GitHub)" <gi...@apache.org> on 2025/01/13 11:01:45 UTC
[D] Dask executor doesn't get correct response if worker restarted during task running [airflow]
GitHub user danxian-baiheng created a discussion: Dask executor doesn't get correct response if worker restarted during task running
### 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
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
GitHub link: https://github.com/apache/airflow/discussions/45608
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] Dask executor doesn't get correct response if worker restarted during task running [airflow]
Posted by "danxian-baiheng (via GitHub)" <gi...@apache.org>.
GitHub user danxian-baiheng added a comment to the discussion: Dask executor doesn't get correct response if worker restarted during task running
I haven't found the DaskExecutor tag in airflow-provider's list.
Does it be removed from airflow in the latest version?
GitHub link: https://github.com/apache/airflow/discussions/45608#discussioncomment-11818792
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] Dask executor doesn't get correct response if worker restarted during task running [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: Dask executor doesn't get correct response if worker restarted during task running
> 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.
GitHub link: https://github.com/apache/airflow/discussions/45608#discussioncomment-11818793
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] Dask executor doesn't get correct response if worker restarted during task running [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
GitHub user boring-cyborg[bot] added a comment to the discussion: Dask executor doesn't get correct response if worker restarted during task running
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.
GitHub link: https://github.com/apache/airflow/discussions/45608#discussioncomment-11818791
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] Dask executor doesn't get correct response if worker restarted during task running [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: Dask executor doesn't get correct response if worker restarted during task running
Converting to a discussion as it's at most that.
GitHub link: https://github.com/apache/airflow/discussions/45608#discussioncomment-11818797
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org