DeciderService { @VisibleForTesting void checkWorkflowTimeout(Workflow workflow) { WorkflowDef workflowDef = workflow.getWorkflowDefinition(); if (workflowDef == null) { LOGGER.warn("Missing workflow definition : {}", workflow.getWorkflowId()); return; } if (workflow.getStatus().isTerminal() || workflowDef.getTimeoutSeconds() <= 0) { return; } long timeout = 1000L * workflowDef.getTimeoutSeconds(); long now = System.currentTimeMillis(); long elapsedTime = workflow.getLastRetriedTime() > 0 ? now - workflow.getLastRetriedTime() : now - workflow.getStartTime(); if (elapsedTime < timeout) { return; } String reason = String.format("Workflow '%s' timed out after %d seconds. Timeout configured as %d. " + "Timeout policy configured to %s", workflow.getWorkflowId(), elapsedTime / 1000L, timeout, workflowDef.getTimeoutPolicy().name()); switch (workflowDef.getTimeoutPolicy()) { case ALERT_ONLY: LOGGER.info(reason); Monitors.recordWorkflowTermination(workflow.getWorkflowName(), WorkflowStatus.TIMED_OUT, workflow.getOwnerApp()); return; case TIME_OUT_WF: throw new TerminateWorkflowException(reason, WorkflowStatus.TIMED_OUT); } } @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 testCheckWorkflowTimeout() { Counter counter = registry.counter("workflow_failure", "class", "WorkflowMonitor", "workflowName", "test", "status", "TIMED_OUT", "ownerApp", "junit"); assertEquals(0, counter.count()); WorkflowDef workflowDef = new WorkflowDef(); workflowDef.setName("test"); Workflow workflow = new Workflow(); workflow.setOwnerApp("junit"); workflow.setStartTime(System.currentTimeMillis() - 10_000); workflow.setWorkflowId("workflow_id"); workflow.setWorkflowDefinition(null); deciderService.checkWorkflowTimeout(workflow); workflow.setWorkflowDefinition(workflowDef); deciderService.checkWorkflowTimeout(workflow); workflowDef.setTimeoutPolicy(WorkflowDef.TimeoutPolicy.ALERT_ONLY); workflowDef.setTimeoutSeconds(2); workflow.setWorkflowDefinition(workflowDef); deciderService.checkWorkflowTimeout(workflow); assertEquals(1, counter.count()); workflowDef.setTimeoutPolicy(WorkflowDef.TimeoutPolicy.TIME_OUT_WF); workflow.setWorkflowDefinition(workflowDef); try { deciderService.checkWorkflowTimeout(workflow); } catch (TerminateWorkflowException twe) { assertTrue(twe.getMessage().contains("Workflow 'workflow_id' timed out")); } workflow.setLastRetriedTime(System.currentTimeMillis() - 5_000); try { deciderService.checkWorkflowTimeout(workflow); } catch (TerminateWorkflowException twe) { assertTrue(twe.getMessage().contains("Workflow 'workflow_id' timed out")); } }