DeciderService { @VisibleForTesting void checkTaskPollTimeout(TaskDef taskDef, Task task) { if (taskDef == null) { LOGGER.warn("Missing task definition for task:{}/{} in workflow:{}", task.getTaskId(), task.getTaskDefName(), task.getWorkflowInstanceId()); return; } if (taskDef.getPollTimeoutSeconds() == null || taskDef.getPollTimeoutSeconds() <= 0 || !task.getStatus().equals(SCHEDULED)) { return; } final long pollTimeout = 1000L * taskDef.getPollTimeoutSeconds(); final long adjustedPollTimeout = pollTimeout + task.getCallbackAfterSeconds() * 1000L; final long now = System.currentTimeMillis(); final long pollElapsedTime = now - (task.getScheduledTime() + ((long) task.getStartDelayInSeconds() * 1000L)); if (pollElapsedTime < adjustedPollTimeout) { return; } String reason = String.format( "Task poll timed out after %d seconds. Poll timeout configured as %d seconds. Timeout policy configured to %s", pollElapsedTime / 1000L, pollTimeout / 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 testCheckTaskPollTimeout() { 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.setPollTimeoutSeconds(1); Task task = new Task(); task.setTaskType(taskType.getName()); task.setScheduledTime(System.currentTimeMillis() - 2_000); task.setStatus(Status.SCHEDULED); deciderService.checkTaskPollTimeout(taskType, task); assertEquals(++counterCount, counter.count()); assertEquals(Status.TIMED_OUT, task.getStatus()); assertNotNull(task.getReasonForIncompletion()); task.setScheduledTime(System.currentTimeMillis()); task.setReasonForIncompletion(null); task.setStatus(Status.SCHEDULED); deciderService.checkTaskPollTimeout(taskType, task); assertEquals(counterCount, counter.count()); assertEquals(Status.SCHEDULED, task.getStatus()); assertNull(task.getReasonForIncompletion()); }