Fix XCom directory creation logic in Kubernetes decorator (#56545)

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

@@ -87,15 +87,20 @@ def __init__(self, namespace: str | None = None, use_dill: bool = False, **kwarg
     def _generate_cmds(self) -> list[str]:
         script_filename = "/tmp/script.py"
         input_filename = "/tmp/script.in"
-        output_filename = "/airflow/xcom/return.json"
+
+        if getattr(self, "do_xcom_push", False):
+            output_filename = "/airflow/xcom/return.json"
+            make_xcom_dir_cmd = "mkdir -p /airflow/xcom"
+        else:
+            output_filename = "/dev/null"
+            make_xcom_dir_cmd = ":"  # shell no-op
 
         write_local_script_file_cmd = (
             f"{_generate_decoded_command(quote(_PYTHON_SCRIPT_ENV), quote(script_filename))}"
         )
         write_local_input_file_cmd = (
             f"{_generate_decoded_command(quote(_PYTHON_INPUT_ENV), quote(input_filename))}"
         )
-        make_xcom_dir_cmd = "mkdir -p /airflow/xcom"
         exec_python_cmd = f"python {script_filename} {input_filename} {output_filename}"
         return [
             "bash",

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/decorators/test_kubernetes.py

@@ -128,3 +128,50 @@ def f(arg1, arg2, kwarg1=None, kwarg2=None):
         # Second container is xcom image
         assert containers[1].image == XCOM_IMAGE
         assert containers[1].volume_mounts[0].mount_path == "/airflow/xcom"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("do_xcom_push", "expected_output", "expects_mkdir"),
+        [
+            (False, "/dev/null", False),
+            (True, "/airflow/xcom/return.json", True),
+        ],
+        ids=["without_xcom_push", "with_xcom_push"],
+    )
+    def test_generated_command_respects_do_xcom_push(
+        self, do_xcom_push: bool, expected_output: str, expects_mkdir: bool
+    ):
+        with self.dag_maker:
+
+            @task.kubernetes(
+                image="python:3.10-slim-buster",
+                in_cluster=False,
+                cluster_context="default",
+                config_file="/tmp/fake_file",
+                namespace="default",
+            )
+            def f():
+                return {"key": "value"}
+
+            k8s_task = f.override(task_id="my_task_id", do_xcom_push=do_xcom_push)()
+
+        if do_xcom_push:
+            self.mock_hook.return_value.get_xcom_sidecar_container_image.return_value = XCOM_IMAGE
+            self.mock_hook.return_value.get_xcom_sidecar_container_resources.return_value = {
+                "requests": {"cpu": "1m", "memory": "10Mi"},
+                "limits": {"cpu": "1m", "memory": "50Mi"},
+            }
+
+        self.execute_task(k8s_task)
+        containers = self.mock_create_pod.call_args.kwargs["pod"].spec.containers
+        assert len(containers) == (2 if do_xcom_push else 1)
+
+        generated_command = containers[0].command
+        assert generated_command
+        assert len(generated_command) >= 3
+
+        bash_command = generated_command[-1]
+        assert expected_output in bash_command
+        assert ("/airflow/xcom" in bash_command) == expects_mkdir
+        if not expects_mkdir:
+            assert " && : && " in bash_command