Fix : Don't treat premature tasks as could_not_run tasks

airflow/jobs.py

@@ -525,7 +525,10 @@ def process_dag(self, dag, queue):
             elif ti.is_runnable(flag_upstream_failed=True):
                 self.logger.debug('Queuing task: {}'.format(ti))
                 queue.put((ti.key, pickle_id))
+            elif ti.is_premature():
+                continue
             else:
+                self.logger.debug('Adding task: {} to the COULD_NOT_RUN set'.format(ti))
                 could_not_run.add(ti)
 
         # this type of deadlock happens when dagruns can't even start and so

airflow/models.py

@@ -856,7 +856,7 @@ def is_queueable(
         if self.execution_date > datetime.now():
             return False
         # is the task still in the retry waiting period?
-        elif self.state == State.UP_FOR_RETRY and not self.ready_for_retry():
+        elif self.is_premature():
             return False
         # does the task have an end_date prior to the execution date?
         elif self.task.end_date and self.execution_date > self.task.end_date:
@@ -878,6 +878,15 @@ def is_queueable(
         else:
             return False
 
+
+    def is_premature(self):
+        """
+        Returns whether a task is in UP_FOR_RETRY state and its retry interval
+        has elapsed.
+        """
+        # is the task still in the retry waiting period?
+        return self.state == State.UP_FOR_RETRY and not self.ready_for_retry()
+
     def is_runnable(
             self,
             include_queued=False,