[EWT-499] Airflow Upgrade to 1.10.12 [CP] from 1.10.4+twtr : 6162402

[TWTTR] Don't enqueue tasks again if already queued for K8sExecutor(twitter-forks#60)

Basically reverting commit 87fcc1c  and making changes specifically into the Celery Executor class only.

airflow/executors/base_executor.py

@@ -56,9 +56,9 @@ def queue_command(self, simple_task_instance, command, priority=1, queue=None):
         key = simple_task_instance.key
         if key not in self.queued_tasks and key not in self.running:
             self.log.info("Adding to queue: %s", command)
+            self.queued_tasks[key] = (command, priority, queue, simple_task_instance)
         else:
-            self.log.info("Adding to queue even though already queued or running {}".format(command, key))
-        self.queued_tasks[key] = (command, priority, queue, simple_task_instance)
+            self.log.error("could not queue task %s", key)
 
     def queue_task_instance(
             self,

airflow/executors/celery_executor.py

@@ -166,6 +166,14 @@ def start(self):
             self._sync_parallelism
         )
 
+    def queue_command(self, simple_task_instance, command, priority=1, queue=None):
+        key = simple_task_instance.key
+        if key not in self.queued_tasks and key not in self.running:
+            self.log.info("Adding to queue: %s", command)
+        else:
+            self.log.info("Adding to queue even though already queued or running {}".format(command, key))
+        self.queued_tasks[key] = (command, priority, queue, simple_task_instance)
+
     def _num_tasks_per_send_process(self, to_send_count):
         """
         How many Celery tasks should each worker process send.

airflow/version.py

@@ -20,3 +20,4 @@
 
 version = '1.10.12'
 
+