This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 72f6173c1b2 Fix KubernetesPodOperator XCom loss when container_logs is
a string (#72502)
72f6173c1b2 is described below
commit 72f6173c1b2a99592d833144c36dc3d9e2bc7229
Author: Henry Chen <[email protected]>
AuthorDate: Wed Sep 9 06:38:48 2026 +0800
Fix KubernetesPodOperator XCom loss when container_logs is a string (#72502)
When container_logs is a single container name, the check that decides
whether the base container still needs to be awaited used `in` on a
string, which is a substring match. Any name containing the base
container name (a typo like "base2", or a sidecar like "base-metrics")
made the operator believe the base container's logs were being followed
and skip the explicit wait. XCom extraction then ran while the base
container was still running, read an empty result, tore down the
sidecar, and the task succeeded with a None XCom.
---
.../providers/cncf/kubernetes/operators/pod.py | 5 ++++-
.../unit/cncf/kubernetes/operators/test_pod.py | 22 ++++++++++++++++++++++
2 files changed, 26 insertions(+), 1 deletion(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
index 5a9df5645aa..1656b8a825e 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -922,8 +922,11 @@ class KubernetesPodOperator(BaseOperator):
container_name_log_prefix_enabled=self.container_name_log_prefix_enabled,
log_formatter=self.log_formatter,
)
+ followed_containers = (
+ [self.container_logs] if isinstance(self.container_logs, str)
else self.container_logs
+ )
if not self.get_logs or (
- self.container_logs is not True and self.base_container_name
not in self.container_logs
+ followed_containers is not True and self.base_container_name
not in followed_containers
):
self.pod_manager.await_container_completion(
pod=pod,
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
index 15a37133c3e..154af0708e2 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
@@ -2126,6 +2126,28 @@ class TestKubernetesPodOperator:
# check that we wait for the xcom sidecar to start before extracting
XCom
mock_await_xcom_sidecar.assert_called_once_with(pod=pod)
+ @pytest.mark.parametrize(
+ ("container_logs", "should_await_base"),
+ [
+ pytest.param("base", False, id="base-as-string"),
+ pytest.param("base2", True,
id="base-is-substring-of-other-container"),
+ ],
+ )
+ @patch(f"{POD_MANAGER_CLASS}.await_container_completion")
+ @patch(f"{POD_MANAGER_CLASS}.fetch_requested_container_logs")
+ def
test_string_container_logs_matches_base_container_by_name_not_substring(
+ self, mock_fetch_log, mock_await_container_completion, container_logs,
should_await_base
+ ):
+ k = KubernetesPodOperator(task_id="task", get_logs=True,
container_logs=container_logs)
+ pod, _ = self.run_pod(k)
+
+ if should_await_base:
+ mock_await_container_completion.assert_called_once_with(
+ pod=pod, container_name="base", polling_time=1
+ )
+ else:
+ mock_await_container_completion.assert_not_called()
+
@patch(HOOK_CLASS, new=MagicMock)
@patch(KUB_OP_PATH.format("find_pod"))
def test_execute_sync_callbacks(self, find_pod_mock):
