Add annotations field into in KubernetesPodOperator (#35641)

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

@@ -191,7 +191,7 @@ class KubernetesPodOperator(BaseOperator):
     :param image_pull_policy: Specify a policy to cache or always pull an image.
     :param annotations: non-identifying metadata you can attach to the Pod.
         Can be a large range of data, and can include characters
-        that are not permitted by labels.
+        that are not permitted by labels. (templated)
     :param container_resources: resources for the launched pod. (templated)
     :param affinity: affinity scheduling rules for the launched pod.
     :param config_file: The path to the Kubernetes config file. (templated)
@@ -261,6 +261,7 @@ class KubernetesPodOperator(BaseOperator):
     template_fields: Sequence[str] = (
         "image",
         "cmds",
+        "annotations",
         "arguments",
         "env_vars",
         "labels",

tests/providers/cncf/kubernetes/operators/test_pod.py

@@ -149,6 +149,7 @@ def test_templates(self, create_task_instance_of_operator):
             arguments="{{ dag.dag_id }}",
             cmds="{{ dag.dag_id }}",
             image="{{ dag.dag_id }}",
+            annotations={"dag-id": "{{ dag.dag_id }}"},
         )
 
         rendered = ti.render_templates()
@@ -167,6 +168,7 @@ def test_templates(self, create_task_instance_of_operator):
         assert dag_id == ti.task.pod_template_file
         assert dag_id == ti.task.arguments
         assert dag_id == ti.task.env_vars[0]
+        assert dag_id == rendered.annotations["dag-id"]
 
     def run_pod(self, operator: KubernetesPodOperator, map_index: int = -1) -> k8s.V1Pod:
         with self.dag_maker(dag_id="dag") as dag: