KPO pods can hang if sidecar never exits

### Apache Airflow Provider(s)

cncf-kubernetes

### Versions of Apache Airflow Providers

all versions

### Apache Airflow version

all versions

### Operating System

linux

### Deployment

Astronomer

### Deployment details

this can happen on any deployment type / executor as it's an issue with KPO.

### What happened

tl;dr KPO sometimes hang around long after their corresponding TI has reached a terminal state.

sometimes the xcom sidecar never exits leaving zombie KPO pods around indefinitely.
this could also happen if the user defines a pod with a sidecar that never exits and ignores signals.

A KubernetesPodOperator (KPO) pod is considered a zombie if any of the following apply:
1. It doesn't match any active (non-terminal) task instance in the database
2. The pod's `try_number` is less than the matching active task instance's current `try_number` (old retry pod)

### What you think should happen instead

There should be some clean-up logic perhaps in the scheduler, instead of the operator clean up itself to force kill zombie KPO pods.

similar to that which cleans up Kubernetes Executor zombies (gated by the configurations below)
AIRFLOW__KUBERNETES_EXECUTOR__DELETE_WORKER_PODS
AIRFLOW__KUBERNETES_EXECUTOR__DELETE_WORKER_PODS_ON_FAILURE

1. **List all KPO pods**: Uses Kubernetes API with label selector `kubernetes_pod_operator=True`
2. **Extract TI details from pods**: For each pod, extracts normalized task instance details from labels (`dag_id`, `task_id`, `run_id`, `map_index`, `try_number`). Pod labels are already normalized by Airflow using `make_safe_label_value()`
3. **Query active task instances**: Queries all active (non-terminal) task instances from the database in a single batch query, filtering for KubernetesPodOperator-related operators (KubernetesPodOperator, KubernetesPodOperatorAsync, SparkKubernetesPodOperator, etc., but excluding "external" EksPodOperator and GKEStartPodOperator)
4. **Normalize DB values to pod label format**: Normalizes database values (`dag_id`, `task_id`, `run_id`, `map_index`) using `make_safe_label_value()` to match the format used in pod labels
5. **Exact matching**: Compares normalized pod labels with normalized database values using exact matching
6. **Identify zombies**: Pods that don't match any active task instance are considered zombies. Pods matching active TIs are checked for old retry scenarios
7. **Terminate zombies**: Force deletes all identified zombie pods

### How to reproduce

I've used this dag for testing this scenario

you can manually kill the zombie pod TIs
```python
from airflow import __version__ as airflow_version
from airflow.decorators import dag
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from kubernetes.client import models as k8s
from pendulum import datetime

def sleep_container(name: str, duration: int | str) -> k8s.V1Container:
return k8s.V1Container(
name=name,
image="busybox:latest",
command=["/bin/sh", "-c"],
args=[f"sleep {duration}"],
)

def healthy_pod_spec() -> k8s.V1PodSpec:
return k8s.V1PodSpec(
containers=[
sleep_container("base", "20"),
],
termination_grace_period_seconds=86400,
restart_policy="Never",
)

def zombie_pod_spec() -> k8s.V1PodSpec:
return k8s.V1PodSpec(
containers=[
sleep_container("base", "5"),
sleep_container("sidecar", "infinity"),
],
)

@dag(
dag_id="kpo_zombie_dag",
schedule=None,
start_date=datetime(2024, 1, 1),
catchup=False,
tags=["test", "kpo", "zombie"],
)
def kpo_zombie_dag():
zombie_task = KubernetesPodOperator(
task_id="zombie_kpo_task",
full_pod_spec=k8s.V1Pod(
metadata=k8s.V1ObjectMeta(),
spec=zombie_pod_spec(),
),
)
healthy_task = KubernetesPodOperator(
task_id="healthy_kpo_task",
full_pod_spec=k8s.V1Pod(
metadata=k8s.V1ObjectMeta(),
spec=healthy_pod_spec(),
),
)

tasks = [zombie_task, healthy_task]

# Create 3 mapped healthy KPO tasks (only for Airflow >= 2.3.0)
# Mapped tasks were introduced in Airflow 2.3.0
try:
# Parse version string (e.g., "2.3.0" or "2.3.0+astro.1")
version_parts = airflow_version.split("+")[0].split(".")
major = int(version_parts[0])
minor = int(version_parts[1]) if len(version_parts) > 1 else 0

supports_mapped_tasks = major > 2 or (major == 2 and minor >= 3)

if supports_mapped_tasks:
# Use expand_kwargs to create 2 mapped task instances (map_index 0, 1)
# one zombie and one healthy
# Each dict in the list creates one mapped instance
mapped_healthy_task = KubernetesPodOperator.partial(
task_id="mapped_kpo_task",
).expand_kwargs(
[
{
"full_pod_spec": k8s.V1Pod(
metadata=k8s.V1ObjectMeta(),
spec=zombie_pod_spec(),
),
},
{
"full_pod_spec": k8s.V1Pod(
metadata=k8s.V1ObjectMeta(),
spec=healthy_pod_spec(),
),
},
]
)
tasks.append(mapped_healthy_task)
except (ValueError, IndexError):
# If version parsing fails, skip mapped tasks to be safe
pass

return tasks

kpo_dag = kpo_zombie_dag()

```

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Start by tracing the scheduler cleanup logic for Kubernetes Executor zombies and the KubernetesPodOperator pod-label and task-instance matching described in the issue. Use the provided kpo_zombie_dag reproduction to verify that orphaned or old-retry KPO pods are force-deleted while pods belonging to active task instances remain.

Written by the indexing model from the issue text.