How can I clean up a DaskExecutor after a flow has finished? I'd like to call

Cluster.stop()

Some Dask clusters for Dask executor have

shutdown_on_finish=True

or something like that. If yours does not, you might have to use a resource manager as seen here

The Cluster class is homegrown and needs to be explicitly stopped. I did try with the resource manager, but found it much harder to work with. Maybe a downstream flow responsible for cleaning up would work?

I guess yeah and then you can do

StartFlowRun(wait=False)

to end the current one, but how would you get the cluster-ip? pass it as a parameter?

The cluster can be retrieved by name rather than IP which makes it easier. I ended up adding the teardown in the terminal_state_handler of the Flow for now.

That’s a great approach!