@Marvin I am using prefect and dask to distribute tasks on Kubernetes. Moreover, there are some heavy tasks which should be run on specific dask workers. If I submit tasks via

dask.annnotate

on a specific worker, and if one of the tasks is failed, I manunally need to interrupt the flow with

cluster.close()

and

client.close()

. It stops the flow, close the connection between client and cluster gracefully, but the prefect UI still shows some task which are running with blue color. I can confirm that the flow is stopped locally and no furter tasks are running or in queue. My question is, how can I make sure that when a task is failed and my own queue mechanism stops the flow, all the running tasks should be stopped gracefully, so the prefect UI gets the updated task run state?