Hi all I'm trying to run a long running task that takes in a local DaskCluster but it seems like something on prefect arbitrarily kills off the worker... My flow uses the Dask ResourceManager looks something like

@task(checkpoint=False)
def long_running_dask_task(inputs: dict, client: Client) -> boolean:
  futures: List[Future] = []

  while(True): 
    next = get_next(inputs)
    if not next:
      break
    f = client.submit(func, next)
    futures.append(f)
    
  client.gather(futures)
  return True
  
with Flow('local-dask-flow') as flow:
  with DaskCluster(...) as client:
    long_running_dask_task(param_1)
  flow.executor = LocalExecutor()

https://docs.prefect.io/core/idioms/resource-manager.html#example-creating-a-temporary-dask-cluster