DeciderService { @VisibleForTesting boolean isResponseTimedOut(TaskDef taskDefinition, Task task) { if (taskDefinition == null) { LOGGER.warn("missing task type : {}, workflowId= {}", task.getTaskDefName(), task.getWorkflowInstanceId()); return false; } if (task.getStatus().isTerminal() || task.getTaskType().equals(SUB_WORKFLOW.name())) { return false; } long now = System.currentTimeMillis(); long callbackTime = 1000L * task.getCallbackAfterSeconds(); long referenceTime = task.getUpdateTime() > 0 ? task.getUpdateTime() : task.getScheduledTime(); long pendingTime = now - (referenceTime + callbackTime); Monitors.recordTaskPendingTime(task.getTaskType(), task.getWorkflowType(), pendingTime); long thresholdMS = config.getIntProperty(PENDING_TASK_TIME_THRESHOLD_PROPERTY_NAME, 60) * 60 * 1000; if (pendingTime > thresholdMS) { LOGGER.warn("Task: {} of type: {} in workflow: {}/{} is in pending state for longer than {} ms", task.getTaskId(), task.getTaskType(), task.getWorkflowInstanceId(), task.getWorkflowType(), thresholdMS); } if (!task.getStatus().equals(IN_PROGRESS) || taskDefinition.getResponseTimeoutSeconds() == 0) { return false; } LOGGER.debug("Evaluating responseTimeOut for Task: {}, with Task Definition: {}", task, taskDefinition); long responseTimeout = 1000L * taskDefinition.getResponseTimeoutSeconds(); long adjustedResponseTimeout = responseTimeout + callbackTime; long noResponseTime = now - task.getUpdateTime(); if (noResponseTime < adjustedResponseTimeout) { LOGGER.debug("Current responseTime: {} has not exceeded the configured responseTimeout of {} for the Task: {} with Task Definition: {}", pendingTime, responseTimeout, task, taskDefinition); return false; } Monitors.recordTaskResponseTimeout(task.getTaskDefName()); return true; } @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);  }
@Test public void testIsResponseTimedOut() { TaskDef taskDef = new TaskDef(); taskDef.setName("test_rt"); taskDef.setResponseTimeoutSeconds(10); Task task = new Task(); task.setTaskDefName("test_rt"); task.setStatus(Status.IN_PROGRESS); task.setTaskId("aa"); task.setTaskType(TaskType.TASK_TYPE_SIMPLE); task.setUpdateTime(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(11)); assertTrue(deciderService.isResponseTimedOut(taskDef, task)); task.setTaskType(TaskType.TASK_TYPE_SUB_WORKFLOW); assertFalse(deciderService.isResponseTimedOut(taskDef, task)); }