Hi, I am running a flow using DaskExecutor, after the Dask cluster starts up successfully, the flow encounters an error and goes FAILED, when I rerun the flow I get the following error.

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"services \"foobar\" already exists","reason":"AlreadyExists","details":{"name":"foobar","kind":"services"},"code":409}

It seems that the error occurs when I try to create a service with the same name. How can I rerun the flow when using DaskExecutor?

Hello! I've never seen this error before. Could you please provide more information about your setup and flow you are running? Are you using any CRDs?

No, I am not using CRDs. I am using the following Pod template.

kind: Pod
metadata:
  labels: {}
spec:
  restartPolicy: Never
  imagePullSecrets:
    - name: gcr-image-puller-service-account
  containers:
    - image: "<http://gcr.io/my-project/prefect:latest-python3.8|gcr.io/my-project/prefect:latest-python3.8>"
      imagePullPolicy: Always
      args: ["dask-worker", "$(DASK_SCHEDULER_ADDRESS)", "--nthreads", "2", "--death-timeout", "60", "--no-dashboard"]
      name: worker
      resources:
        limits:
          cpu: "1"
          memory: "1Gi"
        requests:
          cpu: "1"
          memory: "1Gi"

I have created the DaskExecutor with the following code.

pod = make_pod_from_dict(read_pod_template("MY_POD_TEMPLATE"))
pod.spec.service_account = "MY_SERVICE_ACCOUNT"
executor = DaskExecutor(
    cluster_class=lambda: KubeCluster(
        name="MY_CLUSTER_NAME",
        namespace="MY_NAMESPACE",
        pod_template=pod,
    ),
    adapt_kwargs={"minimum": 1, "maximum": 2},
)