KubernetesJobOperator leaks monitoring pods — on_finish_action is ignored #67332

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.1.8

What happened and how to reproduce it?

Each task run of KubernetesJobOperator leaves behind one (or more, with
parallelism > 1) monitoring pod in the target namespace. Pods accumulate
forever; in our environment we observed 25 orphan pods in a single
namespace after a few weeks of scheduled runs.

The leaked pods are the monitoring/log-streaming pods discovered via
self.get_pods(...) and stored on self.pods. They have no
ownerReferences (they are not owned by the V1Job), so:

• ttl_seconds_after_finished on the Job does not reap them — Job TTL
  only covers Job-owned pods.
• propagation_policy="Foreground" on on_kill does not reap them —
  cascade only follows ownerReferences.
• The on_finish_action parameter inherited from KubernetesPodOperator
  is silently ignored — KubernetesJobOperator.execute() never invokes
  post_complete_action() / cleanup() / process_pod_deletion().

What you think should happen instead?

Same contract as KubernetesPodOperator:

• With default on_finish_action="delete_pod", pods should be deleted
  after the task finishes (success or failure).
• With on_finish_action="delete_succeeded_pod", only successful pods.
• With on_finish_action="keep_pod", pods are kept.

For the Job operator, none of those values currently have any effect on
the monitoring pods.

Operating System

No response

Deployment

Official Apache Airflow Helm Chart

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

1.19.0

Kubernetes Version

1.35.0

Helm Chart configuration

No response

Docker Image customizations

Not applicable

Steps to Reproduce

Prerequisites

• Airflow 3.1.8 with apache-airflow-providers-cncf-kubernetes ≥ 10.x
• A reachable Kubernetes cluster and namespace (repro in the steps below)
• The Airflow scheduler/worker has RBAC pods: get,list,create,delete on that namespace

1. Create a minimal DAG

dags/kjo_leak_repro.py:

from datetime import datetime

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.job import KubernetesJobOperator

with DAG(
    dag_id="kjo_leak_repro",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
    tags=["repro", "kjo-leak"],
) as dag:
    # on_finish_action="delete_pod" is the default and is documented to delete
    # the pod after completion. Set it explicitly to make the contract obvious.
    KubernetesJobOperator(
        task_id="leaky_job",
        namespace="repro",
        image="busybox:1.36",
        cmds=["sh", "-c"],
        arguments=["echo hello && sleep 5"],
        name="kjo-leak-repro",
        on_finish_action="delete_pod",
        wait_until_job_complete=True,
        # ttl reaps the V1Job + its OWNED pods, but NOT the monitoring pod.
        ttl_seconds_after_finished=60,
    )

2. Baseline — confirm namespace is clean

kubectl -n repro get pods -l airflow_kpo_in_cluster=True
# No resources found in repro namespace.

3. Trigger the DAG once

airflow dags trigger kjo_leak_repro

Wait until the task is in success.

4. Observe the leak

After the Job's TTL has elapsed (~60s after completion), the V1Job and its
owned pod are gone, but the monitoring pod remains:

kubectl -n repro get pods -l airflow_kpo_in_cluster=True -o json \
  | jq '[.items[] | {name:.metadata.name, phase:.status.phase,
                     ownerRefs:(.metadata.ownerReferences // [] | map(.kind))}]'

Expected (buggy) output:

[
  {
    "name": "kjo-leak-repro-<hash>",
    "phase": "Succeeded",
    "ownerRefs": []
  }
]

Key indicators that this is the leaked monitoring pod:

• phase: Succeeded — the task itself ran fine.
• ownerReferences: [] — no owner, so neither the Job's TTL controller nor
  propagationPolicy=Foreground on on_kill will ever reap it.
• Labels include both airflow_kpo_in_cluster=True and
  kubernetes_pod_operator=True.

5. Confirm on_finish_action is ignored

Re-run the DAG with each variant; the same orphan monitoring pod appears for
all of them:

on_finish_action="delete_pod"           # leaks
on_finish_action="delete_succeeded_pod" # leaks (succeeded run!)
on_finish_action="keep_pod"             # leaks (same as the other two)

For the Job operator none of these values reach the monitoring pod —
KubernetesJobOperator.execute() never calls post_complete_action() /
cleanup() / process_pod_deletion() (which is what KubernetesPodOperator
uses to honour on_finish_action).

6. Reproduce accumulation (optional)

Trigger the DAG N times:

for i in $(seq 1 10); do airflow dags trigger kjo_leak_repro; sleep 30; done

kubectl -n repro get pods -l airflow_kpo_in_cluster=True \
  -o json | jq '[.items[] | select(.metadata.ownerReferences == null)] | length'
# => 10

One ownerless Succeeded pod per task run — pods grow linearly forever until
something outside Airflow deletes them.

7. Reproduce the on_kill path (optional)

To confirm the same leak on cancellation:

1. Change arguments to ["sleep 600"] so the task runs long enough to kill.
2. Mark the task as failed from the UI (or airflow tasks fail).
3. After the cascade has settled, the monitoring pod is still in repro —
   on_kill's propagationPolicy="Foreground" only follows
   ownerReferences, and the monitoring pod has none.

Expected behaviour (per KubernetesPodOperator contract)

Today, all three rows leak the pod on every run.

Anything else?

• 100 % reproduction rate on every task run.
• Affects every KubernetesJobOperator deployment we are aware of —
  silent because the leak is slow (one pod per run) and the pods are in
  Succeeded state, so they don't trip readiness/liveness alerts.
• Operationally significant on long-running deployments: namespace quotas,
  kubectl get pods latency, and scheduler workload all degrade until
  someone manually cleans up.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct