Fix support for long dag_id and task_id in KubernetesExecutor (#14703)

The key used to remove a task from executor.running is reconstituted
from pod annotations, so make sure the full dag_id and task_id are in
the annotations.

(cherry picked from commit b5e7ada345)

airflow/kubernetes/pod_generator.py

@@ -368,10 +368,6 @@ class PodGenerator:
         except Exception:  # pylint: disable=W0703
             image = kube_image
 
-        task_id = make_safe_label_value(task_id)
-        dag_id = make_safe_label_value(dag_id)
-        scheduler_job_id = make_safe_label_value(str(scheduler_job_id))
-
         dynamic_pod = k8s.V1Pod(
             metadata=k8s.V1ObjectMeta(
                 namespace=namespace,
@@ -383,9 +379,9 @@ class PodGenerator:
                 },
                 name=PodGenerator.make_unique_pod_id(pod_id),
                 labels={
-                    'airflow-worker': scheduler_job_id,
-                    'dag_id': dag_id,
-                    'task_id': task_id,
+                    'airflow-worker': make_safe_label_value(str(scheduler_job_id)),
+                    'dag_id': make_safe_label_value(dag_id),
+                    'task_id': make_safe_label_value(task_id),
                     'execution_date': datetime_to_label_safe_datestring(date),
                     'try_number': str(try_number),
                     'airflow_version': airflow_version.replace('+', '-'),

tests/kubernetes/test_pod_generator.py

@@ -500,6 +500,9 @@ class TestPodGenerator(unittest.TestCase):
         for _, v in result.metadata.labels.items():
             assert len(v) <= 63
 
+        assert 'a' * 512 == result.metadata.annotations['dag_id']
+        assert 'a' * 512 == result.metadata.annotations['task_id']
+
     def test_merge_objects_empty(self):
         annotations = {'foo1': 'bar1'}
         base_obj = k8s.V1ObjectMeta(annotations=annotations)