StateInitializingTrigger implements TriggerListener { @Override public void event(Workflow workflow, Trigger trigger, Instant instant, TriggerParameters parameters) { final var configuration = workflow.configuration(); if (configuration.flyteExecConf().isEmpty() && configuration.dockerImage().isEmpty()) { LOG.warn("{} has no execution execution info, skipping", workflow.id()); return; } final var parameter = toParameter(configuration.schedule(), instant); final var workflowInstance = WorkflowInstance.create(workflow.id(), parameter); try { stateManager.trigger(workflowInstance, trigger, parameters); } catch (IsClosedException isClosedException) { LOG.warn("State receiver is closed when processing workflow {} for trigger {} at {}", workflow, trigger, instant, isClosedException); throw new RuntimeException(isClosedException); } } StateInitializingTrigger(StateManager stateManager); @Override void event(Workflow workflow, Trigger trigger, Instant instant,
                    TriggerParameters parameters);  }
@Test public void shouldInitializeWorkflowInstanceWithoutDockerArgs() throws Exception { WorkflowConfiguration workflowConfiguration = WorkflowConfigurationBuilder.from(dockerWorkflowConfiguration(HOURS)).dockerArgs(Optional.empty()).build(); Workflow workflow = Workflow.create("id", workflowConfiguration); trigger.event(workflow, NATURAL_TRIGGER, TIME, PARAMETERS); WorkflowInstance expectedInstance = WorkflowInstance.create(workflow.id(), toParameter(workflowConfiguration.schedule(), TIME)); verify(stateManager).trigger(expectedInstance, Trigger.natural(), PARAMETERS); }
@Test public void shouldInjectTriggerExecutionEventWithNaturalTrigger() throws Exception { WorkflowConfiguration workflowConfiguration = dockerWorkflowConfiguration(HOURS); Workflow workflow = Workflow.create("id", workflowConfiguration); trigger.event(workflow, NATURAL_TRIGGER, TIME, PARAMETERS); WorkflowInstance expectedInstance = WorkflowInstance.create(workflow.id(), "2016-01-18T09"); verify(stateManager).trigger(expectedInstance, Trigger.natural(), PARAMETERS); }
@Test public void shouldInjectTriggerExecutionEventWithBackfillTrigger() throws Exception { WorkflowConfiguration workflowConfiguration = dockerWorkflowConfiguration(HOURS); Workflow workflow = Workflow.create("id", workflowConfiguration); trigger.event(workflow, BACKFILL_TRIGGER, TIME, PARAMETERS); WorkflowInstance expectedInstance = WorkflowInstance.create(workflow.id(), "2016-01-18T09"); verify(stateManager).trigger(expectedInstance, BACKFILL_TRIGGER, PARAMETERS); }
@Test public void shouldDoNothingIfDockerImageAndFlyteExecConfigIsMissing() { var configuration = WorkflowConfigurationBuilder.from(TestData.MINIMAL_WORKFLOW_CONFIGURATION) .dockerImage(Optional.empty()) .flyteExecConf(Optional.empty()) .build(); var workflow = Workflow.create("id", configuration); trigger.event(workflow, NATURAL_TRIGGER, TIME, PARAMETERS); verifyNoInteractions(stateManager); }
@Test public void shouldCreateWorkflowInstanceParameter() throws Exception { for (Map.Entry<Schedule, String> scheduleCase : SCHEDULE_ARG_EXPECTS.entrySet()) { reset(stateManager); WorkflowConfiguration workflowConfiguration = dockerWorkflowConfiguration( scheduleCase.getKey(), "--date", "{}", "--bar"); Workflow workflow = Workflow.create("id", workflowConfiguration); trigger.event(workflow, NATURAL_TRIGGER, TIME, PARAMETERS); WorkflowInstance expectedInstance = WorkflowInstance.create(workflow.id(), scheduleCase.getValue()); verify(stateManager).trigger(expectedInstance, Trigger.natural(), PARAMETERS); } }
@Test public void shouldWrapIsClosedExceptionIntoRuntime() throws Exception { var cause = new IsClosedException(); doThrow(cause).when(stateManager).trigger(any(), any(), any()); Workflow workflow = Workflow.create("id", dockerWorkflowConfiguration(HOURS)); var exception = assertThrows( RuntimeException.class, () -> trigger.event(workflow, NATURAL_TRIGGER, TIME, PARAMETERS) ); assertThat(exception.getCause(), equalTo(cause)); }