From 501e1dd2634d4c7f98e4dce58d22f4e01c76dd3b Mon Sep 17 00:00:00 2001
From: Dheeren Mohta <dheerenmohta@gmail.com>
Date: Wed, 2 Sep 2026 17:38:16 +0530
Subject: [PATCH] Fix pod name in RenderedTaskInstanceFields for
 KubernetesExecutor

RenderedTaskInstanceFields' k8s pod spec is stamped by the task itself,
after Kubernetes has already created and named the real pod. However,
render_k8s_pod_yaml() rebuilds the pod spec from scratch and assigns it
a brand new random name via create_unique_id(), so the pod name shown
in RTIF's rendered spec almost never matches the pod actually running
the task, confusing anyone trying to correlate the two.

Since we are running inside that real pod by the time RTIF is written,
and Kubernetes sets a pod's hostname to its own metadata.name by
default, socket.gethostname() gives us the real name. Substitute it in
only for that in-pod path, leaving the on-demand preview path (used
when no real pod exists yet) untouched.

Closes: #28186
---
 .../cncf/kubernetes/template_rendering.py     |  9 ++++
 .../kubernetes/test_template_rendering.py     | 41 +++++++++++++++++++
 2 files changed, 50 insertions(+)

diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/template_rendering.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/template_rendering.py
index 6920e8400f2fd..b1b3b1f9b0b6b 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/template_rendering.py
+++ b/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
 
 
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_template_rendering.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_template_rendering.py
index a3fb5d40bc21d..687036b9d513f 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_template_rendering.py
+++ b/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):