Fail the task if the xcom sidecar container does not start for a long time

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

@@ -633,7 +633,9 @@ def execute_sync(self, context: Context):
                 )
 
             if self.do_xcom_push:
-                self.pod_manager.await_xcom_sidecar_container_start(pod=self.pod)
+                self.pod_manager.await_xcom_sidecar_container_start(
+                    pod=self.pod, startup_timeout=self.startup_timeout_seconds
+                )
                 result = self.extract_xcom(pod=self.pod)
             istio_enabled = self.is_istio_enabled(self.pod)
             self.remote_pod = self.pod_manager.await_pod_completion(

airflow/providers/cncf/kubernetes/utils/pod_manager.py

@@ -721,14 +721,27 @@ def read_pod(self, pod: V1Pod) -> V1Pod:
         except HTTPError as e:
             raise AirflowException(f"There was an error reading the kubernetes API: {e}")
 
-    def await_xcom_sidecar_container_start(self, pod: V1Pod) -> None:
+    def await_xcom_sidecar_container_start(self, pod: V1Pod, startup_timeout: int = 120) -> None:
+        """
+        Await xcom sidecar container to start running.
+
+        :param pod:
+        :param startup_timeout:  Timeout (in seconds) for startup of the pod
+        """
         self.log.info("Checking if xcom sidecar container is started.")
+        start_time = time.time()
         for attempt in itertools.count():
             if self.container_is_running(pod, PodDefaults.SIDECAR_CONTAINER_NAME):
                 self.log.info("The xcom sidecar container is started.")
                 break
             if not attempt:
                 self.log.warning("The xcom sidecar container is not yet started.")
+            if time.time() - start_time >= startup_timeout:
+                msg = (
+                    f"Xcom sidecar container took longer than {startup_timeout} seconds to start. "
+                    "Check the container events in kubernetes to determine why."
+                )
+                raise AirflowException(msg)
             time.sleep(1)
 
     def extract_xcom(self, pod: V1Pod) -> str:

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

@@ -1507,7 +1507,7 @@ def test_get_logs_but_not_for_base_container(
         # check that KPO waits for the base container to complete before proceeding to extract XCom
         mock_await_container_completion.assert_called_once_with(pod=pod, container_name="base")
         # check that we wait for the xcom sidecar to start before extracting XCom
-        mock_await_xcom_sidecar.assert_called_once_with(pod=pod)
+        mock_await_xcom_sidecar.assert_called_once_with(pod=pod, startup_timeout=120)
 
     @patch(HOOK_CLASS, new=MagicMock)
     @patch(KUB_OP_PATH.format("find_pod"))