From 135837ae20defa1157ab149992b4c255fd529345 Mon Sep 17 00:00:00 2001 From: manan164 <1897158+manan164@users.noreply.github.com> Date: Mon, 10 Jun 2024 19:57:36 +0530 Subject: [PATCH] fixes --- ...hicalForkJoinSubworkflowRestartSpec.groovy | 32 ++++++------------- ...rchicalForkJoinSubworkflowRetrySpec.groovy | 30 ++++++----------- .../LambdaAndTerminateTaskSpec.groovy | 4 +-- .../NestedForkJoinSubWorkflowSpec.groovy | 16 ++-------- .../integration/SubWorkflowRestartSpec.groovy | 20 +++--------- .../integration/SubWorkflowRetrySpec.groovy | 16 +++------- 6 files changed, 31 insertions(+), 87 deletions(-) diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRestartSpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRestartSpec.groovy index 279a48fb2..b490ba5c7 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRestartSpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRestartSpec.groovy @@ -86,7 +86,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' tasks[2].status == Task.Status.SCHEDULED tasks[3].taskType == TASK_TYPE_JOIN @@ -118,9 +118,9 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.COMPLETED tasks[3].taskType == TASK_TYPE_JOIN tasks[3].status == Task.Status.IN_PROGRESS } @@ -218,7 +218,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' tasks[2].status == Task.Status.SCHEDULED tasks[3].taskType == TASK_TYPE_JOIN @@ -228,11 +228,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { when: "poll and complete the integration_task_2 task in the root workflow" def rootJoinId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTaskByRefName("fanouttask_join").taskId workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) - - and: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newMidLevelWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newMidLevelWorkflowId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new mid level workflow is created and is in RUNNING state" newMidLevelWorkflowId != midLevelWorkflowId @@ -242,9 +238,9 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.COMPLETED tasks[3].taskType == TASK_TYPE_JOIN tasks[3].status == Task.Status.IN_PROGRESS } @@ -252,11 +248,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { when: "poll and complete the integration_task_2 task in the mid-level workflow" def midJoinId = workflowExecutionService.getExecutionStatus(newMidLevelWorkflowId, true).getTaskByRefName("fanouttask_join").taskId workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) - - and: "poll and execute the sub workflow task" - polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(newMidLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId @@ -318,7 +310,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' tasks[2].status == Task.Status.SCHEDULED tasks[3].taskType == TASK_TYPE_JOIN @@ -344,11 +336,7 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification { def midJoinId = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true).getTaskByRefName("fanouttask_join").taskId def rootJoinId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTaskByRefName("fanouttask_join").taskId workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) - - and: "the SUB_WORKFLOW task in mid level workflow is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRetrySpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRetrySpec.groovy index 56d78bfc2..c403b3a14 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRetrySpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRetrySpec.groovy @@ -86,7 +86,7 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' tasks[2].status == Task.Status.SCHEDULED tasks[3].taskType == TASK_TYPE_JOIN @@ -118,19 +118,15 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.COMPLETED tasks[3].taskType == TASK_TYPE_JOIN tasks[3].status == Task.Status.IN_PROGRESS } and: "poll and complete the integration_task_1 task in the mid-level workflow" workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) - - when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) def midLevelWorkflowInstance = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true) then: "verify that the leaf workflow is RUNNING, and first task is in SCHEDULED state" @@ -225,14 +221,12 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { tasks[3].taskType == TASK_TYPE_JOIN tasks[3].status == Task.Status.IN_PROGRESS tasks[4].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[4].status == Task.Status.SCHEDULED + tasks[4].status == Task.Status.IN_PROGRESS tasks[4].retriedTaskId == tasks[1].taskId } when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newMidLevelWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newMidLevelWorkflowId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTasks().get(4).subWorkflowId def rootJoinId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTaskByRefName("fanouttask_join").taskId then: "verify that a new mid level workflow is created and is in RUNNING state" @@ -243,7 +237,7 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { tasks[0].taskType == TASK_TYPE_FORK tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS tasks[2].taskType == 'integration_task_2' tasks[2].status == Task.Status.SCHEDULED tasks[3].taskType == TASK_TYPE_JOIN @@ -253,11 +247,7 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { when: "poll and complete the integration_task_1 task in the mid-level workflow" workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) def midJoinId = workflowExecutionService.getExecutionStatus(newMidLevelWorkflowId, true).getTaskByRefName("fanouttask_join").taskId - - and: "poll and execute the sub workflow task" - polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(newMidLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId @@ -326,7 +316,7 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { tasks[3].taskType == TASK_TYPE_JOIN tasks[3].status == Task.Status.IN_PROGRESS tasks[4].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[4].status == Task.Status.SCHEDULED + tasks[4].status == Task.Status.IN_PROGRESS tasks[4].retriedTaskId == tasks[1].taskId } @@ -346,9 +336,7 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification { } when: "the SUB_WORKFLOW task in mid level workflow is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/LambdaAndTerminateTaskSpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/LambdaAndTerminateTaskSpec.groovy index fd6c6cb14..d6b37890a 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/LambdaAndTerminateTaskSpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/LambdaAndTerminateTaskSpec.groovy @@ -139,7 +139,7 @@ class LambdaAndTerminateTaskSpec extends AbstractSpecification { tasks[3].status == Task.Status.IN_PROGRESS tasks[3].taskType == 'JOIN' tasks[3].seq == 4 - tasks[4].status == Task.Status.SCHEDULED + tasks[4].status == Task.Status.IN_PROGRESS tasks[4].taskType == 'SUB_WORKFLOW' tasks[4].seq == 5 tasks[5].status == Task.Status.IN_PROGRESS @@ -149,8 +149,6 @@ class LambdaAndTerminateTaskSpec extends AbstractSpecification { when: "subworkflow is retrieved" def workflow = workflowExecutionService.getExecutionStatus(workflowInstanceId, true) - def subWorkflowTaskId = workflow.getTaskByRefName("test_terminate_subworkflow").getTaskId() - asyncSystemTaskExecutor.execute(subWorkflowTask, subWorkflowTaskId) workflow = workflowExecutionService.getExecutionStatus(workflowInstanceId, true) def subWorkflowId = workflow.getTaskByRefName("test_terminate_subworkflow").subWorkflowId diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/NestedForkJoinSubWorkflowSpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/NestedForkJoinSubWorkflowSpec.groovy index 548418531..041bfbc67 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/NestedForkJoinSubWorkflowSpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/NestedForkJoinSubWorkflowSpec.groovy @@ -86,7 +86,7 @@ class NestedForkJoinSubWorkflowSpec extends AbstractSpecification { tasks[1].taskType == TASK_TYPE_FORK tasks[1].status == Task.Status.COMPLETED tasks[2].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.IN_PROGRESS tasks[3].taskType == 'integration_task_2' tasks[3].status == Task.Status.SCHEDULED tasks[4].taskType == TASK_TYPE_JOIN @@ -101,10 +101,6 @@ class NestedForkJoinSubWorkflowSpec extends AbstractSpecification { workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task1.integration.worker', ['op': 'task1.done']) workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task1.integration.worker', ['op': 'task1.done']) - and: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - List polledTaskIds = queueDAO.pop("SUB_WORKFLOW", 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds.get(0)) - then: "verify that the 'sub_workflow_task' is in a IN_PROGRESS state" def parentWorkflowInstance = workflowExecutionService.getExecutionStatus(parentWorkflowId, true) with(parentWorkflowInstance) { @@ -314,7 +310,7 @@ class NestedForkJoinSubWorkflowSpec extends AbstractSpecification { tasks[1].taskType == TASK_TYPE_FORK tasks[1].status == Task.Status.COMPLETED tasks[2].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.IN_PROGRESS tasks[3].taskType == 'integration_task_2' tasks[3].status == Task.Status.SCHEDULED tasks[4].taskType == TASK_TYPE_JOIN @@ -326,8 +322,6 @@ class NestedForkJoinSubWorkflowSpec extends AbstractSpecification { } when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - List polledTaskIds = queueDAO.pop("SUB_WORKFLOW", 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds.get(0)) workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done']) def workflow = workflowExecutionService.getExecutionStatus(parentWorkflowId, true) @@ -444,14 +438,10 @@ class NestedForkJoinSubWorkflowSpec extends AbstractSpecification { tasks[6].taskType == TASK_TYPE_JOIN tasks[6].status == Task.Status.IN_PROGRESS tasks[7].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[7].status == Task.Status.SCHEDULED + tasks[7].status == Task.Status.IN_PROGRESS tasks[7].retriedTaskId == tasks[2].taskId } - when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - List polledTaskIds = queueDAO.pop("SUB_WORKFLOW", 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds.get(0)) - then: "verify that SUB_WORKFLOW task in in progress" def parentWorkflowInstance = workflowExecutionService.getExecutionStatus(parentWorkflowId, true) with(workflowExecutionService.getExecutionStatus(parentWorkflowId, true)) { diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRestartSpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRestartSpec.groovy index 519d01a18..88589291d 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRestartSpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRestartSpec.groovy @@ -200,13 +200,9 @@ class SubWorkflowRestartSpec extends AbstractSpecification { tasks[0].taskType == 'integration_task_1' tasks[0].status == Task.Status.COMPLETED tasks[1].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[1].status == Task.Status.SCHEDULED + tasks[1].status == Task.Status.IN_PROGRESS } - - when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newMidLevelWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newMidLevelWorkflowId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new mid level workflow is created and is in RUNNING state" newMidLevelWorkflowId != midLevelWorkflowId @@ -219,11 +215,7 @@ class SubWorkflowRestartSpec extends AbstractSpecification { when: "poll and complete the integration_task_1 task in the mid-level workflow" workflowTestUtil.pollAndCompleteTask('integration_task_1', 'task1.integration.worker', ['op': 'task1.done']) - - and: "poll and execute the sub workflow task" - polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(newMidLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId @@ -307,11 +299,7 @@ class SubWorkflowRestartSpec extends AbstractSpecification { when: "poll and complete the task in the mid level workflow" workflowTestUtil.pollAndCompleteTask('integration_task_1', 'task1.integration.worker', ['op': 'task1.done']) - - and: "the SUB_WORKFLOW task in mid level workflow is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId diff --git a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRetrySpec.groovy b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRetrySpec.groovy index 4cab35a5a..99f882323 100644 --- a/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRetrySpec.groovy +++ b/test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRetrySpec.groovy @@ -198,14 +198,10 @@ class SubWorkflowRetrySpec extends AbstractSpecification { tasks[1].status == Task.Status.FAILED tasks[1].retried tasks[2].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.IN_PROGRESS tasks[2].retriedTaskId == tasks[1].taskId } - - when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newMidLevelWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newMidLevelWorkflowId = workflowExecutionService.getExecutionStatus(rootWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new mid level workflow is created and is in RUNNING state" newMidLevelWorkflowId != midLevelWorkflowId @@ -407,7 +403,7 @@ class SubWorkflowRetrySpec extends AbstractSpecification { tasks[1].status == Task.Status.FAILED tasks[1].retried tasks[2].taskType == TASK_TYPE_SUB_WORKFLOW - tasks[2].status == Task.Status.SCHEDULED + tasks[2].status == Task.Status.IN_PROGRESS tasks[2].retriedTaskId == tasks[1].taskId } @@ -419,11 +415,7 @@ class SubWorkflowRetrySpec extends AbstractSpecification { tasks[1].status == Task.Status.IN_PROGRESS tasks[1].subworkflowChanged } - - when: "the SUB_WORKFLOW task in mid level workflow is started by issuing a system task call" - def polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200) - asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0]) - def newLeafWorkflowId = workflowExecutionService.getTask(polledTaskIds[0]).subWorkflowId + def newLeafWorkflowId = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true).getTasks().get(1).subWorkflowId then: "verify that a new leaf workflow is created and is in RUNNING state" newLeafWorkflowId != leafWorkflowId