Fix KubernetesPodOperator fails to delete pods with None value labels

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

@@ -1208,6 +1208,7 @@ def _build_find_pod_label_selector(self, context: Context | None = None, *, excl
             **self.labels,
             **self._get_ti_pod_labels(context, include_try_number=False),
         }
+        labels = _normalize_labels_dict(labels)
         label_strings = [f"{label_id}={label}" for label_id, label in sorted(labels.items())]
         labels_value = ",".join(label_strings)
         if exclude_checked:
@@ -1280,7 +1281,7 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
             kind="Pod",
             metadata=k8s.V1ObjectMeta(
                 namespace=self.namespace,
-                labels=self.labels,
+                labels=_normalize_labels_dict(self.labels),
                 name=self.name,
                 annotations=self.annotations,
             ),
@@ -1437,3 +1438,8 @@ def __exit__(self, exctype, excinst, exctb) -> bool:
             logger = logging.getLogger(__name__)
             logger.exception(excinst)
         return True
+
+
+def _normalize_labels_dict(labels: dict) -> dict:
+    """Return a copy of the labels dict with all None values replaced by empty strings."""
+    return {k: ("" if v is None else v) for k, v in labels.items()}

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py

@@ -407,7 +407,7 @@ def test_labels(self, hook_mock, in_cluster):
             namespace="default",
             image="ubuntu:16.04",
             cmds=["bash", "-cx"],
-            labels={"foo": "bar"},
+            labels={"foo": "bar", "none_value": None},
             name="test",
             task_id="task",
             in_cluster=in_cluster,
@@ -424,6 +424,7 @@ def test_labels(self, hook_mock, in_cluster):
             "airflow_version": mock.ANY,
             "run_id": "test",
             "airflow_kpo_in_cluster": str(in_cluster),
+            "none_value": "",  # None converted to empty string
         }
 
     def test_labels_mapped(self):
@@ -454,6 +455,49 @@ def test_find_custom_pod_labels(self):
         assert "foo=bar" in label_selector
         assert "hello=airflow" in label_selector
 
+    def test_build_find_pod_label_selector(self):
+        """Comprehensive single test combining all label normalization scenarios.
+
+        Includes: normal labels, None values, empty string, zero, False.
+        Asserts: None -> empty assignment, other falsy preserved, core airflow labels present, no literal 'None'.
+        """
+        k = KubernetesPodOperator(
+            labels={
+                "foo": "bar",
+                "hello": "airflow",
+                "a": None,
+                "c": None,
+                "empty_str": "",
+                "zero": 0,
+                "false": False,
+                "none": None,
+            },
+            name="test",
+            task_id="task",
+        )
+        context = create_context(k)
+        label_selector = k._build_find_pod_label_selector(context)
+
+        # Standard labels
+        assert "foo=bar" in label_selector
+        assert "hello=airflow" in label_selector
+
+        # None normalization (shows as key= with no value)
+        for key in ["a", "c", "none"]:
+            assert f"{key}=" in label_selector
+
+        # Falsy but non-None values preserved verbatim
+        assert "empty_str=" in label_selector
+        assert "zero=0" in label_selector
+        assert "false=False" in label_selector
+
+        # Core Airflow identifying labels always present
+        for core in ["dag_id=dag", "task_id=task", "kubernetes_pod_operator=True", "run_id=test"]:
+            assert core in label_selector
+
+        # Never include literal string 'None'
+        assert "None" not in label_selector
+
     @pytest.mark.asyncio
     @patch(HOOK_CLASS, new=MagicMock)
     def test_find_pod_labels(self):
@@ -1656,6 +1700,27 @@ def test_task_id_as_name_dag_id_is_ignored(self):
         pod = k.build_pod_request_obj({})
         assert re.match(r"a-very-reasonable-task-name-[a-z0-9-]+", pod.metadata.name) is not None
 
+    @pytest.mark.parametrize(
+        "labels,expected",
+        [
+            pytest.param({}, {}, id="empty"),
+            pytest.param(
+                {"a": None, "b": "value", "c": None}, {"a": "", "b": "value", "c": ""}, id="with-none"
+            ),
+            pytest.param(
+                {"empty_str": "", "zero": 0, "false": False, "none": None},
+                {"empty_str": "", "zero": 0, "false": False, "none": ""},
+                id="preserve-other-values",
+            ),
+        ],
+    )
+    def test_normalize_labels_dict(self, labels, expected):
+        """normalize_labels_dict should transform only None values to empty strings and preserve others"""
+        from airflow.providers.cncf.kubernetes.operators.pod import _normalize_labels_dict
+
+        normalized = _normalize_labels_dict(labels)
+        assert normalized == expected
+
     @patch(f"{POD_MANAGER_CLASS}.extract_xcom")
     @patch(f"{POD_MANAGER_CLASS}.await_xcom_sidecar_container_start")
     @patch(f"{POD_MANAGER_CLASS}.await_pod_completion")