DoWhile extends WorkflowSystemTask { @Override @SuppressWarnings("unchecked") public boolean execute(Workflow workflow, Task task, WorkflowExecutor workflowExecutor) { boolean allDone = true; boolean hasFailures = false; StringBuilder failureReason = new StringBuilder(); Map<String, Object> output = new HashMap<>(); task.getOutputData().put("iteration", task.getIteration()); Map<String, Task> relevantTasks = new LinkedHashMap<String, Task>(); Task relevantTask = null; for(Task t : workflow.getTasks()) { if(task.getWorkflowTask().has(TaskUtils.removeIterationFromTaskRefName(t.getReferenceTaskName())) && !task.getReferenceTaskName().equals(t.getReferenceTaskName())) { relevantTask = relevantTasks.get(t.getReferenceTaskName()); if(relevantTask == null || t.getRetryCount() > relevantTask.getRetryCount()) { relevantTasks.put(t.getReferenceTaskName(), t); } } } Collection<Task> loopOver = relevantTasks.values(); for (Task loopOverTask : loopOver) { Status taskStatus = loopOverTask.getStatus(); hasFailures = !taskStatus.isSuccessful(); if (hasFailures) { failureReason.append(loopOverTask.getReasonForIncompletion()).append(" "); } output.put(TaskUtils.removeIterationFromTaskRefName(loopOverTask.getReferenceTaskName()), loopOverTask.getOutputData()); allDone = taskStatus.isTerminal(); if (!allDone || hasFailures) { break; } } task.getOutputData().put(String.valueOf(task.getIteration()), output); if (hasFailures) { logger.debug("taskid {} failed in {} iteration", task.getTaskId(), task.getIteration() + 1); return updateLoopTask(task, Status.FAILED, failureReason.toString()); } else if (!allDone) { return false; } boolean shouldContinue; try { shouldContinue = getEvaluatedCondition(workflow, task, workflowExecutor); logger.debug("taskid {} condition evaluated to {}", task.getTaskId(), shouldContinue); if (shouldContinue) { task.setIteration(task.getIteration() + 1); return scheduleNextIteration(task, workflow, workflowExecutor); } else { logger.debug("taskid {} took {} iterations to complete", task.getTaskId(), task.getIteration() + 1); return markLoopTaskSuccess(task); } } catch (ScriptException e) { String message = String.format("Unable to evaluate condition %s , exception %s", task.getWorkflowTask().getLoopCondition(), e.getMessage()); logger.error(message); logger.error("Marking task {} failed with error.", task.getTaskId()); return updateLoopTask(task, Status.FAILED_WITH_TERMINAL_ERROR, message); } } DoWhile(); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor executor); @Override @SuppressWarnings("unchecked") boolean execute(Workflow workflow, Task task, WorkflowExecutor workflowExecutor);  }
@Test public void testSingleFailedIteration() { task1.setStatus(Task.Status.FAILED); String reason = "Test"; task1.setReasonForIncompletion(reason); Mockito.doReturn(Arrays.asList(task1, task2, loopTask)).when(workflow).getTasks(); boolean success = doWhile.execute(workflow, loopTask, provider); Assert.assertTrue(success); Assert.assertEquals(loopTask.getStatus(), Task.Status.FAILED); Assert.assertFalse(reason.equals(loopTask.getReasonForIncompletion())); }
@Test public void testInProgress() { loopTask.setStatus(Task.Status.IN_PROGRESS); task1.setStatus(Task.Status.IN_PROGRESS); Mockito.doReturn(Arrays.asList(task1, task2, loopTask)).when(workflow).getTasks(); boolean success = doWhile.execute(workflow, loopTask, provider); Assert.assertFalse(success); Assert.assertTrue(loopTask.getStatus() == Task.Status.IN_PROGRESS); }