airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bo...@apache.org
Subject incubator-airflow git commit: [AIRFLOW-1050] Do not count up_for_retry as not ready
Date Fri, 07 Apr 2017 06:00:31 GMT
Repository: incubator-airflow
Updated Branches:
  refs/heads/v1-8-test ebfc3ea73 -> 0fa593e38


[AIRFLOW-1050] Do not count up_for_retry as not ready

up_for_retry tasks were incorrectly counted
towards not_ready
therefore marking a dag run deadlocked instead of
retrying.

Closes #2225 from bolkedebruin/AIRFLOW-1050

(cherry picked from commit 35e43f5067f4741640278b765c0e54e4fd45ffa3)
Signed-off-by: Bolke de Bruin <bolke@xs4all.nl>


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/0fa593e3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/0fa593e3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/0fa593e3

Branch: refs/heads/v1-8-test
Commit: 0fa593e38c7ea88765408af10abad3c3780ba27d
Parents: ebfc3ea
Author: Bolke de Bruin <bolke@xs4all.nl>
Authored: Fri Apr 7 08:00:10 2017 +0200
Committer: Bolke de Bruin <bolke@xs4all.nl>
Committed: Fri Apr 7 08:00:23 2017 +0200

----------------------------------------------------------------------
 airflow/jobs.py | 9 +++++++++
 1 file changed, 9 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/0fa593e3/airflow/jobs.py
----------------------------------------------------------------------
diff --git a/airflow/jobs.py b/airflow/jobs.py
index ce45e05..11ff926 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -1925,6 +1925,15 @@ class BackfillJob(BaseJob):
                             started.pop(key)
                         continue
 
+                    # special case
+                    if ti.state == State.UP_FOR_RETRY:
+                        self.logger.debug("Task instance {} retry period not expired yet"
+                                          .format(ti))
+                        if key in started:
+                            started.pop(key)
+                        tasks_to_run[key] = ti
+                        continue
+
                     # all remaining tasks
                     self.logger.debug('Adding {} to not_ready'.format(ti))
                     not_ready.add(key)


Mime
View raw message