DeciderService { @VisibleForTesting Optional<Task> retry(TaskDef taskDefinition, WorkflowTask workflowTask, Task task, Workflow workflow) throws TerminateWorkflowException { int retryCount = task.getRetryCount(); if (taskDefinition == null) { taskDefinition = metadataDAO.getTaskDef(task.getTaskDefName()); } final int expectedRetryCount = taskDefinition == null ? 0 : Optional.ofNullable(workflowTask).map(WorkflowTask::getRetryCount).orElse(taskDefinition.getRetryCount()); if (!task.getStatus().isRetriable() || SystemTaskType.isBuiltIn(task.getTaskType()) || expectedRetryCount <= retryCount) { if (workflowTask != null && workflowTask.isOptional()) { return Optional.empty(); } WorkflowStatus status = task.getStatus().equals(TIMED_OUT) ? WorkflowStatus.TIMED_OUT : WorkflowStatus.FAILED; updateWorkflowOutput(workflow, task); throw new TerminateWorkflowException(task.getReasonForIncompletion(), status, task); } int startDelay = taskDefinition.getRetryDelaySeconds(); switch (taskDefinition.getRetryLogic()) { case FIXED: startDelay = taskDefinition.getRetryDelaySeconds(); break; case EXPONENTIAL_BACKOFF: int retryDelaySeconds = taskDefinition.getRetryDelaySeconds() * (int) Math.pow(2, task.getRetryCount()); startDelay = retryDelaySeconds < 0 ? Integer.MAX_VALUE : retryDelaySeconds; break; } task.setRetried(true); Task rescheduled = task.copy(); rescheduled.setStartDelayInSeconds(startDelay); rescheduled.setCallbackAfterSeconds(startDelay); rescheduled.setRetryCount(task.getRetryCount() + 1); rescheduled.setRetried(false); rescheduled.setTaskId(IDGenerator.generate()); rescheduled.setRetriedTaskId(task.getTaskId()); rescheduled.setStatus(SCHEDULED); rescheduled.setPollCount(0); rescheduled.setInputData(new HashMap<>()); rescheduled.getInputData().putAll(task.getInputData()); rescheduled.setReasonForIncompletion(null); rescheduled.setSubWorkflowId(null); if (StringUtils.isNotBlank(task.getExternalInputPayloadStoragePath())) { rescheduled.setExternalInputPayloadStoragePath(task.getExternalInputPayloadStoragePath()); } else { rescheduled.getInputData().putAll(task.getInputData()); } if (workflowTask != null && workflow.getSchemaVersion() > 1) { Workflow workflowInstance = populateWorkflowAndTaskData(workflow); Map<String, Object> taskInput = parametersUtils.getTaskInputV2(workflowTask.getInputParameters(), workflowInstance, rescheduled.getTaskId(), taskDefinition); rescheduled.getInputData().putAll(taskInput); } externalizeTaskData(rescheduled); return Optional.of(rescheduled); } @Inject  DeciderService(ParametersUtils parametersUtils, MetadataDAO metadataDAO,
                          ExternalPayloadStorageUtils externalPayloadStorageUtils,
                          @Named("TaskMappers") Map<String, TaskMapper> taskMappers, Configuration configuration); DeciderOutcome decide(Workflow workflow); List<Task> getTasksToBeScheduled(Workflow workflow,
                                            WorkflowTask taskToSchedule, int retryCount); List<Task> getTasksToBeScheduled(Workflow workflow,
                                            WorkflowTask taskToSchedule, int retryCount, String retriedTaskId);  }
@SuppressWarnings("unchecked") @Test public void testTaskRetry() { Workflow workflow = createDefaultWorkflow(); workflow.getWorkflowDefinition().setSchemaVersion(2); Map<String, Object> inputParams = new HashMap<>(); inputParams.put("workflowInputParam", "${workflow.input.requestId}"); inputParams.put("taskOutputParam", "${task2.output.location}"); inputParams.put("constParam", "Some String value"); inputParams.put("nullValue", null); inputParams.put("task2Status", "${task2.status}"); inputParams.put("null", null); inputParams.put("task_id", "${CPEWF_TASK_ID}"); Map<String, Object> env = new HashMap<>(); env.put("env_task_id", "${CPEWF_TASK_ID}"); inputParams.put("env", env); Map<String, Object> taskInput = parametersUtils.getTaskInput(inputParams, workflow, null, "t1"); Task task = new Task(); task.getInputData().putAll(taskInput); task.setStatus(Status.FAILED); task.setTaskId("t1"); TaskDef taskDef = new TaskDef(); WorkflowTask workflowTask = new WorkflowTask(); workflowTask.getInputParameters().put("task_id", "${CPEWF_TASK_ID}"); workflowTask.getInputParameters().put("env", env); Optional<Task> task2 = deciderService.retry(taskDef, workflowTask, task, workflow); assertEquals("t1", task.getInputData().get("task_id")); assertEquals("t1", ((Map<String, Object>) task.getInputData().get("env")).get("env_task_id")); assertNotSame(task.getTaskId(), task2.get().getTaskId()); assertEquals(task2.get().getTaskId(), task2.get().getInputData().get("task_id")); assertEquals(task2.get().getTaskId(), ((Map<String, Object>) task2.get().getInputData().get("env")).get("env_task_id")); Task task3 = new Task(); task3.getInputData().putAll(taskInput); task3.setStatus(Status.FAILED_WITH_TERMINAL_ERROR); task3.setTaskId("t1"); when(metadataDAO.getWorkflowDef(anyString(), anyInt())).thenReturn(Optional.of(new WorkflowDef())); exception.expect(TerminateWorkflowException.class); deciderService.retry(taskDef, workflowTask, task3, workflow); }
@Test public void testWorkflowTaskRetry() { Workflow workflow = createDefaultWorkflow(); workflow.getWorkflowDefinition().setSchemaVersion(2); Map<String, Object> inputParams = new HashMap<>(); inputParams.put("workflowInputParam", "${workflow.input.requestId}"); inputParams.put("taskOutputParam", "${task2.output.location}"); inputParams.put("constParam", "Some String value"); inputParams.put("nullValue", null); inputParams.put("task2Status", "${task2.status}"); inputParams.put("null", null); inputParams.put("task_id", "${CPEWF_TASK_ID}"); Map<String, Object> env = new HashMap<>(); env.put("env_task_id", "${CPEWF_TASK_ID}"); inputParams.put("env", env); Map<String, Object> taskInput = parametersUtils.getTaskInput(inputParams, workflow, null, "t1"); Task task = new Task(); task.getInputData().putAll(taskInput); task.setStatus(Status.FAILED); task.setTaskId("t1"); TaskDef taskDef = new TaskDef(); assertEquals(3, taskDef.getRetryCount()); WorkflowTask workflowTask = new WorkflowTask(); workflowTask.getInputParameters().put("task_id", "${CPEWF_TASK_ID}"); workflowTask.getInputParameters().put("env", env); workflowTask.setRetryCount(1); Optional<Task> task2 = deciderService.retry(taskDef, workflowTask, task, workflow); assertEquals("t1", task.getInputData().get("task_id")); assertEquals("t1", ((Map<String, Object>) task.getInputData().get("env")).get("env_task_id")); assertNotSame(task.getTaskId(), task2.get().getTaskId()); assertEquals(task2.get().getTaskId(), task2.get().getInputData().get("task_id")); assertEquals(task2.get().getTaskId(), ((Map<String, Object>) task2.get().getInputData().get("env")).get("env_task_id")); task2.get().setStatus(Status.FAILED); exception.expect(TerminateWorkflowException.class); final Optional<Task> task3 = deciderService.retry(taskDef, workflowTask, task2.get(), workflow); assertFalse(task3.isPresent()); assertEquals(WorkflowStatus.FAILED, workflow.getStatus()); }
@Test public void testExponentialBackoff() { Workflow workflow = createDefaultWorkflow(); Task task = new Task(); task.setStatus(Status.FAILED); task.setTaskId("t1"); TaskDef taskDef = new TaskDef(); taskDef.setRetryDelaySeconds(60); taskDef.setRetryLogic(TaskDef.RetryLogic.EXPONENTIAL_BACKOFF); WorkflowTask workflowTask = new WorkflowTask(); Optional<Task> task2 = deciderService.retry(taskDef, workflowTask, task, workflow); assertEquals(60, task2.get().getCallbackAfterSeconds()); Optional<Task> task3 = deciderService.retry(taskDef, workflowTask, task2.get(), workflow); assertEquals(120, task3.get().getCallbackAfterSeconds()); Optional<Task> task4 = deciderService.retry(taskDef, workflowTask, task3.get(), workflow); assertEquals(240, task4.get().getCallbackAfterSeconds()); taskDef.setRetryCount(Integer.MAX_VALUE); task4.get().setRetryCount(Integer.MAX_VALUE - 100); Optional<Task> task5 = deciderService.retry(taskDef, workflowTask, task4.get(), workflow); assertEquals(Integer.MAX_VALUE, task5.get().getCallbackAfterSeconds()); }