@VisibleForTesting
    Optional<TaskModel> retry(
            TaskDef taskDefinition,
            WorkflowTask workflowTask,
            TaskModel task,
            WorkflowModel 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()
                || TaskType.isBuiltIn(task.getTaskType())
                || expectedRetryCount <= retryCount) {
            if (workflowTask != null && workflowTask.isOptional()) {
                return Optional.empty();
            }
            WorkflowModel.Status status;
            switch (task.getStatus()) {
                case CANCELED:
                    status = WorkflowModel.Status.TERMINATED;
                    break;
                case TIMED_OUT:
                    status = WorkflowModel.Status.TIMED_OUT;
                    break;
                default:
                    status = WorkflowModel.Status.FAILED;
                    break;
            }
            updateWorkflowOutput(workflow, task);
            final String errMsg =
                    String.format(
                            "Task %s failed with status: %s and reason: '%s'",
                            task.getTaskId(), status, task.getReasonForIncompletion());
            throw new TerminateWorkflowException(errMsg, status, task);
        }

        // retry... - but not immediately - put a delay...
        int startDelay = taskDefinition.getRetryDelaySeconds();
        switch (taskDefinition.getRetryLogic()) {
            case FIXED:
                startDelay = taskDefinition.getRetryDelaySeconds();
                break;
            case LINEAR_BACKOFF:
                int linearRetryDelaySeconds =
                        taskDefinition.getRetryDelaySeconds()
                                * taskDefinition.getBackoffScaleFactor()
                                * (task.getRetryCount() + 1);
                // Reset integer overflow to max value
                startDelay =
                        linearRetryDelaySeconds < 0 ? Integer.MAX_VALUE : linearRetryDelaySeconds;
                break;
            case EXPONENTIAL_BACKOFF:
                int exponentialRetryDelaySeconds =
                        taskDefinition.getRetryDelaySeconds()
                                * (int) Math.pow(2, task.getRetryCount());
                // Reset integer overflow to max value
                startDelay =
                        exponentialRetryDelaySeconds < 0
                                ? Integer.MAX_VALUE
                                : exponentialRetryDelaySeconds;
                break;
        }

        task.setRetried(true);

        TaskModel 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<>(task.getInputData()));
        rescheduled.setReasonForIncompletion(null);
        rescheduled.setSubWorkflowId(null);
        rescheduled.setSeq(0);
        rescheduled.setScheduledTime(0);
        rescheduled.setStartTime(0);
        rescheduled.setEndTime(0);
        rescheduled.setWorkerId(null);

        if (StringUtils.isNotBlank(task.getExternalInputPayloadStoragePath())) {
            rescheduled.setExternalInputPayloadStoragePath(
                    task.getExternalInputPayloadStoragePath());
        } else {
            rescheduled.addInput(task.getInputData());
        }
        if (workflowTask != null && workflow.getWorkflowDefinition().getSchemaVersion() > 1) {
            Map<String, Object> taskInput =
                    parametersUtils.getTaskInputV2(
                            workflowTask.getInputParameters(),
                            workflow,
                            rescheduled.getTaskId(),
                            taskDefinition);
            rescheduled.addInput(taskInput);
        }
        // for the schema version 1, we do not have to recompute the inputs
        return Optional.of(rescheduled);
    }
Updates the workflow output.

@param workflow the workflow instance
@param task if not null, the output of this task will be copied to workflow output if no
    output parameters are specified in the workflow definition if null, the output of the
    last task in the workflow will be copied to workflow output of no output parameters are
    specified in the workflow definition
core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java