Fix bug with retries (default 1)

Only CWLStepOperator is set to retry
Barski-lab · Aug 30, 2019 · 2af52e4 · 2af52e4
1 parent 36ed42b
commit 2af52e4
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 7 deletions.
diff --git a/cwl_airflow/cwldag.py b/cwl_airflow/cwldag.py
@@ -48,7 +48,7 @@ def __init__(
             'tmp_folder': conf_get_default('cwl', 'tmp_folder', '/tmp'),
             'basedir': conf_get_default('cwl', 'tmp_folder', '/tmp'),
             'no_match_user': conf_get_default('cwl', 'no_match_user', False),
-            'task_retries': conf_get_default('cwl', 'retry', 0),
+            'task_retries': conf_get_default('cwl', 'retry', 1),
             'quiet': False,
             'strict': False,
             'on_error': 'continue',
@@ -73,13 +73,15 @@ def create(self):
         if self.cwlwf["class"] in ["CommandLineTool", "ExpressionTool"]:
             cwl_task = CWLStepOperator(task_id=self.dag_id,
                                        dag=self,
+                                       retries=self.default_args["task_retries"],
                                        ui_color='#5C6BC0')
         else:
             outputs = {}
 
             for step_id, step_val in self.cwlwf["steps"].items():
                 cwl_task = CWLStepOperator(task_id=step_id,
                                            dag=self,
+                                           retries=self.default_args["task_retries"],
                                            ui_color='#5C6BC0')
                 outputs[step_id] = cwl_task
 

diff --git a/cwl_airflow/cwlstepoperator.py b/cwl_airflow/cwlstepoperator.py
@@ -52,8 +52,7 @@ def __init__(
 
         kwargs.update({"on_success_callback": kwargs.get("on_success_callback", task_on_success),
                        "on_failure_callback": kwargs.get("on_failure_callback", task_on_failure),
-                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry),
-                       "retries":             kwargs.get("retries", self.dag.default_args["task_retries"])})
+                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry)})
 
         super(self.__class__, self).__init__(task_id=task_id, *args, **kwargs)
 

diff --git a/cwl_airflow/operators/cwljobdispatcher.py b/cwl_airflow/operators/cwljobdispatcher.py
@@ -35,8 +35,7 @@ def __init__(
 
         kwargs.update({"on_success_callback": kwargs.get("on_success_callback", task_on_success),
                        "on_failure_callback": kwargs.get("on_failure_callback", task_on_failure),
-                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry),
-                       "retries":             kwargs.get("retries", self.dag.default_args["task_retries"])})
+                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry)})
 
         super(CWLJobDispatcher, self).__init__(task_id=task_id, *args, **kwargs)
 

diff --git a/cwl_airflow/operators/cwljobgatherer.py b/cwl_airflow/operators/cwljobgatherer.py
@@ -30,8 +30,7 @@ def __init__(
 
         kwargs.update({"on_success_callback": kwargs.get("on_success_callback", task_on_success),
                        "on_failure_callback": kwargs.get("on_failure_callback", task_on_failure),
-                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry),
-                       "retries":             kwargs.get("retries", self.dag.default_args["task_retries"])})
+                       "on_retry_callback":   kwargs.get("on_retry_callback",   task_on_retry)})
 
         super(CWLJobGatherer, self).__init__(task_id=task_id, *args, **kwargs)