Use the real pod hostname in RenderedTaskInstanceFields for the Kubernetes executor

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

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+import os
+import socket
 from typing import TYPE_CHECKING
 
 from jinja2 import TemplateAssertionError, UndefinedError
@@ -82,6 +84,13 @@ def render_k8s_pod_yaml(task_instance: TaskInstance) -> dict | None:
         with_mutation_hook=True,
     )
     sanitized_pod = ApiClient().sanitize_for_serialization(pod)
+    if os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD"):
+        # We are running inside the pod Kubernetes actually created for this task, so we
+        # know its real name: by default Kubernetes sets a pod's hostname to its own
+        # metadata.name. Use that instead of the pod_id above, which is a freshly
+        # regenerated create_unique_id() value and therefore never matches the pod that
+        # is actually running (see GH#28186).
+        sanitized_pod.setdefault("metadata", {})["name"] = socket.gethostname()
     return sanitized_pod
 
 

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_template_rendering.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import os
+import socket
 from unittest import mock
 
 import pytest
@@ -105,6 +106,46 @@ def test_render_k8s_pod_yaml(pod_mutation_hook, create_task_instance):
     pod_mutation_hook.assert_called_once_with(mock.ANY)
 
 
+@mock.patch.dict(os.environ, {"AIRFLOW_IS_K8S_EXECUTOR_POD": "True"})
+@mock.patch("airflow.settings.pod_mutation_hook")
+def test_render_k8s_pod_yaml_uses_real_pod_name_inside_k8s_executor_pod(
+    pod_mutation_hook, create_task_instance
+):
+    """The rendered pod name must match the pod actually running the task (GH#28186).
+
+    Inside a KubernetesExecutor pod, render_k8s_pod_yaml() otherwise regenerates a fresh,
+    random pod_id via create_unique_id() that never matches the real pod Kubernetes already
+    created. Kubernetes sets a pod's own hostname to its metadata.name by default, so the
+    real name is recoverable from within the running pod.
+    """
+    ti = create_task_instance(
+        dag_id="test_render_k8s_pod_yaml_real_name",
+        run_id="test_run_id",
+        task_id="op1",
+        logical_date=DEFAULT_DATE,
+    )
+
+    assert render_k8s_pod_yaml(ti)["metadata"]["name"] == socket.gethostname()
+
+
+@mock.patch("airflow.settings.pod_mutation_hook")
+def test_render_k8s_pod_yaml_keeps_generated_name_outside_k8s_executor_pod(
+    pod_mutation_hook, create_task_instance
+):
+    """Outside a real pod (e.g. the on-demand preview path), no real pod exists yet, so the
+    freshly generated placeholder name must be kept rather than substituted with our own
+    (unrelated) hostname.
+    """
+    ti = create_task_instance(
+        dag_id="test_render_k8s_pod_yaml_preview_name",
+        run_id="test_run_id",
+        task_id="op1",
+        logical_date=DEFAULT_DATE,
+    )
+
+    assert render_k8s_pod_yaml(ti)["metadata"]["name"] != socket.gethostname()
+
+
 @mock.patch.dict(os.environ, {"AIRFLOW_IS_K8S_EXECUTOR_POD": "True"})
 @mock.patch("airflow.settings.pod_mutation_hook")
 def test_render_k8s_pod_yaml_with_custom_pod_template(pod_mutation_hook, create_task_instance, tmp_path):