Fixing queue prioritization

airflow/jobs.py

@@ -283,11 +283,9 @@ class SchedulerJob(BaseJob):
                         logging.debug('Triggering retry: ' + str(ti))
                         executor.queue_task_instance(ti)
                 elif ti.state == State.QUEUED:
-                    # If task instance if up for retry, make sure
+                    # If was queued we skipped so that in gets prioritized
-                    if ti.is_runnable():
+                    # in self.prioritize_queued
-                        logging.debug(
+                    continue
-                            'Starting previously queued : ' + str(ti))
-                        executor.queue_task_instance(ti)
                 else:
                     # Trying to run the next schedule
                     next_schedule = (