public void updateTask(TaskResult taskResult) {
        if (taskResult == null) {
            LOGGER.info("null task given for update");
            throw new ApplicationException(Code.INVALID_INPUT, "Task object is null");
        }

        String workflowId = taskResult.getWorkflowInstanceId();
        Workflow workflowInstance = executionDAOFacade.getWorkflowById(workflowId, true);

        // FIXME Backwards compatibility for legacy workflows already running.
        // This code will be removed in a future version.
        if (workflowInstance.getWorkflowDefinition() == null) {
            workflowInstance = metadataMapperService.populateWorkflowWithDefinitions(workflowInstance);
        }

        Task task = Optional.ofNullable(executionDAOFacade.getTaskById(taskResult.getTaskId()))
                .orElseThrow(() -> new ApplicationException(Code.NOT_FOUND, "No such task found by id: " + taskResult.getTaskId()));

        LOGGER.debug("Task: {} belonging to Workflow {} being updated", task, workflowInstance);

        String taskQueueName = QueueUtils.getQueueName(task);
        if (workflowInstance.getStatus().isTerminal()) {
            // Workflow is in terminal state
            queueDAO.remove(taskQueueName, taskResult.getTaskId());
            LOGGER.info("Workflow: {} has already finished execution. Task update for: {} ignored and removed from Queue: {}.", workflowInstance, taskResult.getTaskId(), taskQueueName);
            Monitors.recordUpdateConflict(task.getTaskType(), workflowInstance.getWorkflowName(), workflowInstance.getStatus());
            return;
        }

        if (task.getStatus().isTerminal()) {
            // Task was already updated....
            queueDAO.remove(taskQueueName, taskResult.getTaskId());
            LOGGER.info("Task: {} has already finished execution with status:{} at {} within workflow: {}. Removed task from queue: {}", task.getTaskId(), task.getStatus(), task.getEndTime(), workflowInstance.getWorkflowId(), taskQueueName);
            Monitors.recordUpdateConflict(task.getTaskType(), workflowInstance.getWorkflowName(), task.getStatus());
            return;
        }

        task.setStatus(valueOf(taskResult.getStatus().name()));
        task.setOutputMessage(taskResult.getOutputMessage());
        task.setReasonForIncompletion(taskResult.getReasonForIncompletion());
        task.setWorkerId(taskResult.getWorkerId());
        task.setCallbackAfterSeconds(taskResult.getCallbackAfterSeconds());

        task.setOutputData(taskResult.getOutputData());
        if (task.getOutputData() != null) {
            externalPayloadStorageUtils.verifyAndUpload(task, TASK_OUTPUT);
        } else {
            task.setExternalOutputPayloadStoragePath(taskResult.getExternalOutputPayloadStoragePath());
        }

        if (task.getStatus().isTerminal()) {
            task.setEndTime(System.currentTimeMillis());
        }

        executionDAOFacade.updateTask(task);

        //If the task has failed update the failed task reference name in the workflow.
        //This gives the ability to look at workflow and see what tasks have failed at a high level.
        if (FAILED.equals(task.getStatus()) || FAILED_WITH_TERMINAL_ERROR.equals(task.getStatus())) {
            workflowInstance.getFailedReferenceTaskNames().add(task.getReferenceTaskName());
            executionDAOFacade.updateWorkflow(workflowInstance);
            LOGGER.debug("Task: {} has a {} status and the Workflow has been updated with failed task reference", task, task.getStatus());
        }

        taskResult.getLogs().forEach(taskExecLog -> taskExecLog.setTaskId(task.getTaskId()));
        executionDAOFacade.addTaskExecLog(taskResult.getLogs());

        switch (task.getStatus()) {
            case COMPLETED:
            case CANCELED:
            case FAILED:
            case FAILED_WITH_TERMINAL_ERROR:
            case TIMED_OUT:
                queueDAO.remove(taskQueueName, taskResult.getTaskId());
                LOGGER.debug("Task: {} removed from taskQueue: {} since the task status is {}", task, taskQueueName, task.getStatus().name());
                break;
            case IN_PROGRESS:
                // put it back in queue based on callbackAfterSeconds
                long callBack = taskResult.getCallbackAfterSeconds();
                queueDAO.remove(taskQueueName, task.getTaskId());
                LOGGER.debug("Task: {} removed from taskQueue: {} since the task status is {}", task, taskQueueName, task.getStatus().name());

                queueDAO.push(taskQueueName, task.getTaskId(), callBack); // Milliseconds
                LOGGER.debug("Task: {} pushed back to taskQueue: {} since the task status is {} with callbackAfterSeconds: {}", task, taskQueueName, task.getStatus().name(), callBack);
                break;
            default:
                break;
        }

        decide(workflowId);

        if (task.getStatus().isTerminal()) {
            long duration = getTaskDuration(0, task);
            long lastDuration = task.getEndTime() - task.getStartTime();
            Monitors.recordTaskExecutionTime(task.getTaskDefName(), duration, true, task.getStatus());
            Monitors.recordTaskExecutionTime(task.getTaskDefName(), lastDuration, false, task.getStatus());
        }
    }
public void updateTask ( TaskResult taskResult ) { if ( taskResult == null ) { LOGGER . info ( "null task given for update" ) ; throw new ApplicationException ( Code . INVALID_INPUT , "Task object is null" ) ; } String workflowId = taskResult . getWorkflowInstanceId ( ) ; Workflow workflowInstance = executionDAOFacade . getWorkflowById ( workflowId , true ) ; if ( workflowInstance . getWorkflowDefinition ( ) == null ) { workflowInstance = metadataMapperService . populateWorkflowWithDefinitions ( workflowInstance ) ; } Task task = Optional . ofNullable ( executionDAOFacade . getTaskById ( taskResult . getTaskId ( ) ) ) . orElseThrow ( ( ) -> new ApplicationException ( Code . NOT_FOUND , "No such task found by id: " + taskResult . getTaskId ( ) ) ) ; LOGGER . debug ( "Task: {} belonging to Workflow {} being updated" , task , workflowInstance ) ; String taskQueueName = QueueUtils . getQueueName ( task ) ; if ( workflowInstance . getStatus ( ) . isTerminal ( ) ) { queueDAO . remove ( taskQueueName , taskResult . getTaskId ( ) ) ; LOGGER . info ( "Workflow: {} has already finished execution. Task update for: {} ignored and removed from Queue: {}." , workflowInstance , taskResult . getTaskId ( ) , taskQueueName ) ; Monitors . recordUpdateConflict ( task . getTaskType ( ) , workflowInstance . getWorkflowName ( ) , workflowInstance . getStatus ( ) ) ; return ; } if ( task . getStatus ( ) . isTerminal ( ) ) { queueDAO . remove ( taskQueueName , taskResult . getTaskId ( ) ) ; LOGGER . info ( "Task: {} has already finished execution with status:{} at {} within workflow: {}. Removed task from queue: {}" , task . getTaskId ( ) , task . getStatus ( ) , task . getEndTime ( ) , workflowInstance . getWorkflowId ( ) , taskQueueName ) ; Monitors . recordUpdateConflict ( task . getTaskType ( ) , workflowInstance . getWorkflowName ( ) , task . getStatus ( ) ) ; return ; } task . setStatus ( valueOf ( taskResult . getStatus ( ) . name ( ) ) ) ; task . setOutputMessage ( taskResult . getOutputMessage ( ) ) ; task . setReasonForIncompletion ( taskResult . getReasonForIncompletion ( ) ) ; task . setWorkerId ( taskResult . getWorkerId ( ) ) ; task . setCallbackAfterSeconds ( taskResult . getCallbackAfterSeconds ( ) ) ; task . setOutputData ( taskResult . getOutputData ( ) ) ; if ( task . getOutputData ( ) != null ) { externalPayloadStorageUtils . verifyAndUpload ( task , TASK_OUTPUT ) ; } else { task . setExternalOutputPayloadStoragePath ( taskResult . getExternalOutputPayloadStoragePath ( ) ) ; } if ( task . getStatus ( ) . isTerminal ( ) ) { task . setEndTime ( System . currentTimeMillis ( ) ) ; } executionDAOFacade . updateTask ( task ) ; if ( FAILED . equals ( task . getStatus ( ) ) || FAILED_WITH_TERMINAL_ERROR . equals ( task . getStatus ( ) ) ) { workflowInstance . getFailedReferenceTaskNames ( ) . add ( task . getReferenceTaskName ( ) ) ; executionDAOFacade . updateWorkflow ( workflowInstance ) ; LOGGER . debug ( "Task: {} has a {} status and the Workflow has been updated with failed task reference" , task , task . getStatus ( ) ) ; } taskResult . getLogs ( ) . forEach ( taskExecLog -> taskExecLog . setTaskId ( task . getTaskId ( ) ) ) ; executionDAOFacade . addTaskExecLog ( taskResult . getLogs ( ) ) ; switch ( task . getStatus ( ) ) { case COMPLETED : case CANCELED : case FAILED : case FAILED_WITH_TERMINAL_ERROR : case TIMED_OUT : queueDAO . remove ( taskQueueName , taskResult . getTaskId ( ) ) ; LOGGER . debug ( "Task: {} removed from taskQueue: {} since the task status is {}" , task , taskQueueName , task . getStatus ( ) . name ( ) ) ; break ; case IN_PROGRESS : long callBack = taskResult . getCallbackAfterSeconds ( ) ; queueDAO . remove ( taskQueueName , task . getTaskId ( ) ) ; LOGGER . debug ( "Task: {} removed from taskQueue: {} since the task status is {}" , task , taskQueueName , task . getStatus ( ) . name ( ) ) ; queueDAO . push ( taskQueueName , task . getTaskId ( ) , callBack ) ; LOGGER . debug ( "Task: {} pushed back to taskQueue: {} since the task status is {} with callbackAfterSeconds: {}" , task , taskQueueName , task . getStatus ( ) . name ( ) , callBack ) ; break ; default : break ; } decide ( workflowId ) ; if ( task . getStatus ( ) . isTerminal ( ) ) { long duration = getTaskDuration ( 0 , task ) ; long lastDuration = task . getEndTime ( ) - task . getStartTime ( ) ; Monitors . recordTaskExecutionTime ( task . getTaskDefName ( ) , duration , true , task . getStatus ( ) ) ; Monitors . recordTaskExecutionTime ( task . getTaskDefName ( ) , lastDuration , false , task . getStatus ( ) ) ; } }
@param taskResult the task result to be updated
@throws ApplicationException