Move pod_mutation_hook call from PodManager to KubernetesPodOperator (#…

…20596)

Previously, in KubernetesPodOperator, the invocation of the pod mutation hook occurred
within the call to PodManager.run_pod_async.  So, `run_pod_async` would not quite run
the pod you asked it to run, but would mutate it first.

With this change, `run_pod_async` runs exactly the pod you request, and the pod returned
by `build_pod_request_obj` is actually the pod you request.

airflow/providers/cncf/kubernetes/CHANGELOG.rst

@@ -27,6 +27,8 @@ Breaking changes
 
 * ``Simplify KubernetesPodOperator (#19572)``
 * Class ``pod_launcher.PodLauncher`` renamed to ``pod_manager.PodManager``
+* :func:`airflow.settings.pod_mutation_hook` is no longer called in :meth:`~cncf.kubernetes.utils.pod_manager.PodManager.run_pod_async``. For ``KubernetesPodOperator``, mutation now occurs in ``build_pod_request_obj``.
+
 
 .. warning:: Many methods in :class:`~.KubernetesPodOperator` and class:`~.PodManager` (formerly named ``PodLauncher``)
     have been renamed. If you have subclassed :class:`~.KubernetesPodOperator` you will need to update your subclass to

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

@@ -25,6 +25,7 @@
 from kubernetes.client import CoreV1Api, models as k8s
 
 from airflow.providers.cncf.kubernetes.utils.pod_manager import PodLaunchFailedException, PodManager, PodPhase
+from airflow.settings import pod_mutation_hook
 
 try:
     import airflow.utils.yaml as yaml
@@ -574,6 +575,7 @@ def build_pod_request_obj(self, context=None):
                 'kubernetes_pod_operator': 'True',
             }
         )
+        pod_mutation_hook(pod)
         return pod
 
 

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

@@ -36,7 +36,6 @@
 from airflow.exceptions import AirflowException
 from airflow.kubernetes.kube_client import get_kube_client
 from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
 from airflow.utils.log.logging_mixin import LoggingMixin
 
 
@@ -104,8 +103,6 @@ def __init__(
 
     def run_pod_async(self, pod: V1Pod, **kwargs) -> V1Pod:
         """Runs POD asynchronously"""
-        pod_mutation_hook(pod)
-
         sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
         json_pod = json.dumps(sanitized_pod, indent=2)