DeciderService { @VisibleForTesting void checkTaskTimeout(TaskDef taskDef, Task task) { if (taskDef == null) { LOGGER.warn("Missing task definition for task:{}/{} in workflow:{}", task.getTaskId(), task.getTaskDefName(), task.getWorkflowInstanceId()); return; } if (task.getStatus().isTerminal() || taskDef.getTimeoutSeconds() <= 0 || task.getStartTime() <= 0) { return; } long timeout = 1000L * taskDef.getTimeoutSeconds(); long now = System.currentTimeMillis(); long elapsedTime = now - (task.getStartTime() + ((long) task.getStartDelayInSeconds() * 1000L)); if (elapsedTime < timeout) { return; } String reason = String.format("Task timed out after %d seconds. Timeout configured as %d seconds. " + "Timeout policy configured to %s", elapsedTime / 1000L, timeout / 1000L, taskDef.getTimeoutPolicy().name()); timeoutTaskWithTimeoutPolicy(reason, taskDef, task); } @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 testTaskTimeout() { Counter counter = registry.counter("task_timeout", "class", "WorkflowMonitor", "taskType", "test"); long counterCount = counter.count(); TaskDef taskType = new TaskDef(); taskType.setName("test"); taskType.setTimeoutPolicy(TimeoutPolicy.RETRY); taskType.setTimeoutSeconds(1); Task task = new Task(); task.setTaskType(taskType.getName()); task.setStartTime(System.currentTimeMillis() - 2_000); task.setStatus(Status.IN_PROGRESS); deciderService.checkTaskTimeout(taskType, task); assertEquals(Status.TIMED_OUT, task.getStatus()); assertNotNull(task.getReasonForIncompletion()); assertEquals(++counterCount, counter.count()); taskType.setTimeoutPolicy(TimeoutPolicy.ALERT_ONLY); task.setStatus(Status.IN_PROGRESS); task.setReasonForIncompletion(null); deciderService.checkTaskTimeout(taskType, task); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertNull(task.getReasonForIncompletion()); assertEquals(++counterCount, counter.count()); boolean exception = false; taskType.setTimeoutPolicy(TimeoutPolicy.TIME_OUT_WF); task.setStatus(Status.IN_PROGRESS); task.setReasonForIncompletion(null); try { deciderService.checkTaskTimeout(taskType, task); } catch (TerminateWorkflowException tw) { exception = true; } assertTrue(exception); assertEquals(Status.TIMED_OUT, task.getStatus()); assertNotNull(task.getReasonForIncompletion()); assertEquals(++counterCount, counter.count()); taskType.setTimeoutPolicy(TimeoutPolicy.TIME_OUT_WF); task.setStatus(Status.IN_PROGRESS); task.setReasonForIncompletion(null); deciderService.checkTaskTimeout(null, task); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertNull(task.getReasonForIncompletion()); assertEquals(counterCount, counter.count()); }