CP 87fcc1c: [TWTR][CX-17516] Requeue tasks in the queued state (twitt…

…er-forks#27)
vshshjn7 · Mar 11, 2020 · 08c4bf4 · 08c4bf4
1 parent fd3d7e5
commit 08c4bf4
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py
@@ -57,9 +57,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("could not queue task %s", key)
+            self.log.info("Adding to queue even though already queued or running {}".format(command, key))
+        self.queued_tasks[key] = (command, priority, queue, task_instance)
 
     def queue_task_instance(
             self,

diff --git a/airflow/jobs.py b/airflow/jobs.py
@@ -1708,8 +1708,13 @@ def _execute_helper(self, processor_manager):
                 scheduled_dag_ids = ", ".join(simple_dag_bag.dag_ids)
                 self.log.info('DAGs to be executed: {}'.format(scheduled_dag_ids))
 
+                # TODO(CX-17516): State.QUEUED has been added here which is a hack as the Celery
+                # Executor does not reliably enqueue tasks with the my MySQL broker, and we have
+                # seen tasks hang after they get queued. The effect of this hack is queued tasks
+                # will constantly be requeued and resent to the executor (Celery).
+                # This should be removed when we switch away from the MySQL Celery backend.
                 self._execute_task_instances(simple_dag_bag,
-                                             (State.SCHEDULED,))
+                                             (State.SCHEDULED, State.QUEUED))
 
             # Call heartbeats
             self.log.debug("Heartbeating the executor")