Retries shouldn't execute if join task failed because of child task

Change-Id: Ideaa9938497f74335af633044cb6e98fbb1522d8
Closes-Bug: #1819418
Signed-off-by: Oleg Ovcharuk <vgvoleg@gmail.com>

mistral/engine/policies.py

@@ -160,6 +160,15 @@ def _ensure_context_has_key(runtime_context, key):
     return runtime_context
 
 
+def _has_incomplete_inbound_tasks(task_ex):
+    if "triggered_by" not in task_ex.runtime_context:
+        return False
+    for trigger in task_ex.runtime_context["triggered_by"]:
+        if trigger["event"] == "not triggered":
+            return True
+    return False
+
+
 class WaitBeforePolicy(base.TaskPolicy):
     _schema = {
         "properties": {
@@ -381,6 +390,11 @@ class RetryPolicy(base.TaskPolicy):
             (self._continue_on_clause and not continue_on_evaluation)
         )
 
+        stop_continue_flag = (
+            stop_continue_flag or
+            _has_incomplete_inbound_tasks(task_ex)
+        )
+
         break_triggered = (
             task_ex.state == states.ERROR and
             break_on_evaluation

mistral/tests/unit/engine/test_policies.py

@@ -1227,6 +1227,49 @@ class PoliciesTest(base.EngineTestCase):
 
         self.assertDictEqual({'result': 'mocked result'}, wf_output)
 
+    def test_retry_failed_join_task(self):
+        retry_wb = """---
+        version: '2.0'
+
+        name: wb
+
+        workflows:
+          wf1:
+            task-defaults:
+              retry:
+                count: 1
+                delay: 0
+            tasks:
+              task1:
+                action: std.noop
+                on-success: join_task
+              task2:
+                action: std.fail
+                on-success: join_task
+              join_task:
+                action: std.noop
+                join: all
+        """
+        wb_service.create_workbook_v2(retry_wb)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('wb.wf1')
+
+        self.await_workflow_error(wf_ex.id)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            tasks = wf_ex.task_executions
+
+        self._assert_single_item(
+            tasks, name="task2", state=states.ERROR
+        )
+        self._assert_single_item(
+            tasks, name="join_task", state=states.ERROR
+        )
+
     @mock.patch.object(
         std_actions.EchoAction,
         'run',