DaskJob stuck in Running forever

[guozhans]

Describe the issue:
handle_runner_status_change_succeeded in dask_kubernetes/operator/controller/controller.py is not idempotent, so a transient API error in the middle of the handler permanently wedges the DaskJob in jobStatus: Running even though the job runner pod completed successfully or badly

The handler does, in order:
DaskCluster.get(name) + cluster.delete()
DaskJob.get(name) + job.patch({"status": {"jobStatus": "Successful", ...}})

What we observed

• The job-runner pod reached Succeeded; the handler fired and deleted the DaskCluster (the DELETE returned 200).

• The very next API call (DaskJob.get) failed with a transient 429 Too Many Requests from the API server, so the handler raised and kopf scheduled a retry.

• Every retry re-runs the handler from the top: DaskCluster.get() now raises kr8s.NotFoundError (the cluster was already deleted in step 1), so the handler fails forever. In our case kopf logged 690+ retries over 13 hours.

• The DaskJob status is never patched, so it stays Running permanently. Anything that watches the DaskJob status to determine completion (in our case the Flyte dask plugin) waits forever.

handle_runner_status_change_failed has the identical delete-then-patch structure and is affected the same way.

Traceback from the operator logs (repeats on every retry):

kopf.objects [ERROR] [<namespace>/<job-name>-runner] Handler 'handle_runner_status_change_succeeded/status.phase' failed with an exception. Will retry.
Traceback (most recent call last):
  ...
  File ".../dask_kubernetes/operator/controller/controller.py", line 861, in handle_runner_status_change_succeeded
    cluster = await DaskCluster.get(name, namespace=namespace)
  File ".../kr8s/_objects.py", line 246, in get
    return await cls.async_get(
  File ".../kr8s/_objects.py", line 318, in async_get
    raise NotFoundError(
kr8s._exceptions.NotFoundError: Could not find DaskCluster <cluster-name> in namespace <namespace>.

kopf progress annotation on the runner pod confirming the retry loop:

'handle_runner_status_change_succeeded/status.phase': {
  'started': '...T12:20:00', 'delayed': '...', 'purpose': 'update',
  'retries': 690, 'success': False, 'failure': False,
  'message': 'Could not find DaskCluster <cluster-name> in namespace <namespace>.'
}

Timeline from the operator log for the affected job:

12:20:00 kopf.objects [INFO ] [<ns>/<job>-runner] Job succeeded, deleting Dask cluster.
12:20:01 httpx        [INFO ] HTTP Request: DELETE .../daskclusters/<cluster> "HTTP/1.1 200 OK"
12:20:0x httpx.HTTPStatusError: Client error '429 Too Many Requests' for url '.../daskjobs?...fieldSelector=metadata.name=<cluster>'
12:21:01 kopf.objects [INFO ] [<ns>/<job>-runner] Job succeeded, deleting Dask cluster.   <- retry
         kr8s._exceptions.NotFoundError: Could not find DaskCluster <cluster> ...          <- fails forever
... (repeats every ~60s indefinitely)

Make the handler idempotent by tolerating an already-deleted cluster, so a retry can still complete the status patch:

    name = meta["labels"]["dask.org/cluster-name"]
    try:
        cluster = await DaskCluster.get(name, namespace=namespace)
        await cluster.delete()
    except kr8s.NotFoundError:
        logger.info("Dask cluster already deleted.")
    job = await DaskJob.get(name, namespace=namespace)
    await job.patch({"status": {"jobStatus": "Successful", ...}}, subresource="status")

Minimal Complete Verifiable Example:
The trigger is a race (any transient API failure between the cluster delete and the job status patch), so it's hard to reproduce deterministically end-to-end, but the broken invariant is easy to demonstrate: any re-invocation of handle_runner_status_change_succeeded after the DaskCluster is gone raises NotFoundError instead of completing the status patch. E.g. create a DaskJob, let the runner succeed, and delete the DaskCluster out of band before/while the handler runs (or inject a fault between the two API calls) — the DaskJob then stays Running forever.

Anything else we need to know?:

Environment:

• Dask kubernetes version: 2025.7.0 (helm chart dask-kubernetes-operator-2025.7.0, image ghcr.io/dask/dask-kubernetes-operator:2025.7.0, reports dask-kubernetes 2025.7.1.dev0+g1daa175)
• kopf: 1.38.0, kr8s: 0.20.8, httpx: 0.28.1
• Kubernetes version: v1.34.8 (managed AKS)
• Install method: helm