Fix a bug where Kube config "worker_pod_pending_fatal_container_state_reasons" is parsed wrongly

providers/cncf/kubernetes/provider.yaml

@@ -241,8 +241,8 @@ config:
         version_added: 8.1.0
         type: string
         example: ~
-        default: 'CreateContainerConfigError,ErrImagePull,CreateContainerError,ImageInspectError,
-        InvalidImageName'
+        default: "CreateContainerConfigError,ErrImagePull,CreateContainerError,ImageInspectError,\
+        InvalidImageName"
       worker_pods_creation_batch_size:
         description: |
           Number of Kubernetes Worker Pod creation calls per scheduler loop.

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py

@@ -169,7 +169,7 @@ def get_provider_info():
                         "version_added": "8.1.0",
                         "type": "string",
                         "example": None,
-                        "default": "CreateContainerConfigError,ErrImagePull,CreateContainerError,ImageInspectError, InvalidImageName",
+                        "default": "CreateContainerConfigError,ErrImagePull,CreateContainerError,ImageInspectError,InvalidImageName",
                     },
                     "worker_pods_creation_batch_size": {
                         "description": 'Number of Kubernetes Worker Pod creation calls per scheduler loop.\nNote that the current default of "1" will only launch a single pod\nper-heartbeat. It is HIGHLY recommended that users increase this\nnumber to match the tolerance of their kubernetes cluster for\nbetter performance.\n',

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_config.py

@@ -42,9 +42,12 @@ def __init__(self):
         )
         self.worker_pod_pending_fatal_container_state_reasons = []
         if conf.get(self.kubernetes_section, "worker_pod_pending_fatal_container_state_reasons", fallback=""):
-            self.worker_pod_pending_fatal_container_state_reasons = conf.get(
-                self.kubernetes_section, "worker_pod_pending_fatal_container_state_reasons"
-            ).split(",")
+            self.worker_pod_pending_fatal_container_state_reasons = [
+                r.strip()
+                for r in conf.get(
+                    self.kubernetes_section, "worker_pod_pending_fatal_container_state_reasons"
+                ).split(",")
+            ]
 
         self.worker_pods_creation_batch_size = conf.getint(
             self.kubernetes_section, "worker_pods_creation_batch_size"

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -1806,3 +1806,24 @@ def effect():
                 self.watcher.run()
 
             mock_underscore_run.assert_called_once_with(mock.ANY, "0", mock.ANY, mock.ANY)
+
+    @pytest.mark.parametrize(
+        "state_reasons, expected_result",
+        [
+            pytest.param("e1,e2,e3", ["e1", "e2", "e3"]),
+            pytest.param("e1, e2,e3", ["e1", "e2", "e3"]),
+            pytest.param(" e1,e2, e3", ["e1", "e2", "e3"]),
+            pytest.param("e1", ["e1"]),
+            pytest.param("e1 ", ["e1"]),
+        ],
+    )
+    def test_kube_config_parse_worker_pod_pending_fatal_container_state_reasons(
+        self, state_reasons, expected_result
+    ):
+        config = {
+            ("kubernetes_executor", "worker_pod_pending_fatal_container_state_reasons"): state_reasons,
+        }
+        with conf_vars(config):
+            executor = KubernetesExecutor()
+
+        assert executor.kube_config.worker_pod_pending_fatal_container_state_reasons == expected_result