@Test
    public void testRetryFromLastFailedSubWorkflowTaskThenStartWithLastFailedTask() {
        IDGenerator idGenerator = new IDGenerator();
        // given
        String id = idGenerator.generate();
        String workflowInstanceId = idGenerator.generate();
        TaskModel task = new TaskModel();
        task.setTaskType(TaskType.SIMPLE.name());
        task.setTaskDefName("task");
        task.setReferenceTaskName("task_ref");
        task.setWorkflowInstanceId(workflowInstanceId);
        task.setScheduledTime(System.currentTimeMillis());
        task.setTaskId(idGenerator.generate());
        task.setStatus(TaskModel.Status.COMPLETED);
        task.setRetryCount(0);
        task.setWorkflowTask(new WorkflowTask());
        task.setOutputData(new HashMap<>());
        task.setSubWorkflowId(id);
        task.setSeq(1);

        TaskModel task1 = new TaskModel();
        task1.setTaskType(TaskType.SIMPLE.name());
        task1.setTaskDefName("task1");
        task1.setReferenceTaskName("task1_ref");
        task1.setWorkflowInstanceId(workflowInstanceId);
        task1.setScheduledTime(System.currentTimeMillis());
        task1.setTaskId(idGenerator.generate());
        task1.setStatus(TaskModel.Status.FAILED);
        task1.setRetryCount(0);
        task1.setWorkflowTask(new WorkflowTask());
        task1.setOutputData(new HashMap<>());
        task1.setSubWorkflowId(id);
        task1.setSeq(2);

        WorkflowModel subWorkflow = new WorkflowModel();
        subWorkflow.setWorkflowId(id);
        WorkflowDef workflowDef = new WorkflowDef();
        workflowDef.setName("subworkflow");
        workflowDef.setVersion(1);
        subWorkflow.setWorkflowDefinition(workflowDef);
        subWorkflow.setStatus(WorkflowModel.Status.FAILED);
        subWorkflow.getTasks().addAll(Arrays.asList(task, task1));
        subWorkflow.setParentWorkflowId("testRunWorkflowId");

        TaskModel task2 = new TaskModel();
        task2.setWorkflowInstanceId(subWorkflow.getWorkflowId());
        task2.setScheduledTime(System.currentTimeMillis());
        task2.setTaskId(idGenerator.generate());
        task2.setStatus(TaskModel.Status.FAILED);
        task2.setRetryCount(0);
        task2.setOutputData(new HashMap<>());
        task2.setSubWorkflowId(id);
        task2.setTaskType(TaskType.SUB_WORKFLOW.name());

        WorkflowModel workflow = new WorkflowModel();
        workflow.setWorkflowId("testRunWorkflowId");
        workflow.setStatus(WorkflowModel.Status.FAILED);
        workflow.setTasks(Collections.singletonList(task2));
        workflowDef = new WorkflowDef();
        workflowDef.setName("first_workflow");
        workflow.setWorkflowDefinition(workflowDef);

        // when
        when(executionDAOFacade.getWorkflowModel(workflow.getWorkflowId(), true))
                .thenReturn(workflow);
        when(executionDAOFacade.getWorkflowModel(task.getSubWorkflowId(), true))
                .thenReturn(subWorkflow);
        when(metadataDAO.getWorkflowDef(anyString(), anyInt()))
                .thenReturn(Optional.of(workflowDef));
        when(executionDAOFacade.getTaskModel(subWorkflow.getParentWorkflowTaskId()))
                .thenReturn(task1);
        when(executionDAOFacade.getWorkflowModel(subWorkflow.getParentWorkflowId(), false))
                .thenReturn(workflow);

        workflowExecutor.retry(workflow.getWorkflowId(), true);

        // then
        assertEquals(task.getStatus(), TaskModel.Status.COMPLETED);
        assertEquals(task1.getStatus(), TaskModel.Status.IN_PROGRESS);
        assertEquals(workflow.getPreviousStatus(), WorkflowModel.Status.FAILED);
        assertEquals(workflow.getStatus(), WorkflowModel.Status.RUNNING);
        assertEquals(subWorkflow.getPreviousStatus(), WorkflowModel.Status.FAILED);
        assertEquals(subWorkflow.getStatus(), WorkflowModel.Status.RUNNING);
    }
Simulate Queue push failures and assert that scheduleTask doesn't throw an exception.
core/src/test/java/com/netflix/conductor/core/execution/TestWorkflowExecutor.java