TerminationHandler implements OutputHandler { @Override public void transitionInto(RunState state, EventRouter eventRouter) { switch (state.state()) { case TERMINATED: if (state.data().lastExit().map(v -> v.equals(0)).orElse(false)) { eventRouter.receiveIgnoreClosed(Event.success(state.workflowInstance()), state.counter()); } else { checkRetry(state, eventRouter); } break; case FAILED: checkRetry(state, eventRouter); break; default: } } TerminationHandler(RetryUtil retryUtil, Supplier<Map<WorkflowId, Workflow>> workflows); @Override void transitionInto(RunState state, EventRouter eventRouter); static final int MISSING_DEPS_RETRY_DELAY_MINUTES; }
@Test public void shouldCompleteOnZeroExitCode() { StateData data = data(1, 1.0, Optional.of(0)); RunState zeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); terminationHandler.transitionInto(zeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.success(WORKFLOW_INSTANCE), COUNTER); }
@Test public void shouldScheduleRetryOnNonZero() { StateData data = data(1, 1.0, Optional.of(1)); RunState nonZeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); Duration expectedDelay = Duration.ofMillis(4711); when(retryUtil.calculateDelay(anyInt())).thenReturn(expectedDelay); terminationHandler.transitionInto(nonZeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.retryAfter(WORKFLOW_INSTANCE, expectedDelay.toMillis()), COUNTER); }
@Test public void shouldScheduleRetryOnMissingExitCode() { StateData data = data(1, 1.0, Optional.empty()); RunState nonZeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); Duration expectedDelay = Duration.ofMillis(4711); when(retryUtil.calculateDelay(anyInt())).thenReturn(expectedDelay); terminationHandler.transitionInto(nonZeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.retryAfter(WORKFLOW_INSTANCE, expectedDelay.toMillis()), COUNTER); }
@Test public void shouldScheduleRetryOnFail() { StateData data = data(1, 1.0, Optional.of(1)); RunState failed = RunState.create(WORKFLOW_INSTANCE, FAILED, data, NOW, COUNTER); Duration expectedDelay = Duration.ofMillis(4711); when(retryUtil.calculateDelay(anyInt())).thenReturn(expectedDelay); terminationHandler.transitionInto(failed, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.retryAfter(WORKFLOW_INSTANCE, expectedDelay.toMillis()), COUNTER); }
@Test public void shouldStopOnNonZeroMaxRetriesReached() { StateData data = data(400, MAX_RETRY_COST, Optional.of(1)); RunState maxedTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); terminationHandler.transitionInto(maxedTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.stop(WORKFLOW_INSTANCE), COUNTER); }
@Test public void shouldStopOnFailMaxRetriesReached() { StateData data = data(400, MAX_RETRY_COST, Optional.of(1)); RunState maxedTerm = RunState.create(WORKFLOW_INSTANCE, FAILED, data, NOW, COUNTER); terminationHandler.transitionInto(maxedTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.stop(WORKFLOW_INSTANCE), COUNTER); }
@Test public void shouldScheduleRetryOf10MinutesOnMissingDependencies() { StateData data = data(1, 1.0, Optional.of(20)); RunState missingDeps = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); terminationHandler.transitionInto(missingDeps, eventRouter); verify(eventRouter).receiveIgnoreClosed( Event.retryAfter(WORKFLOW_INSTANCE, Duration.ofMinutes(10).toMillis()), COUNTER); }
@Test public void shouldStopOnFailFastExitCodeReceived() { var data = data(1, 1.0, Optional.of(50)); var failed = RunState.create(WORKFLOW_INSTANCE, FAILED, data, NOW, COUNTER); terminationHandler.transitionInto(failed, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.stop(WORKFLOW_INSTANCE), COUNTER); }
@Test public void shouldNotStopOnWorkflowNotFound() { when(workflows.get()).thenReturn(Map.of()); var data = data(1, 1.0, Optional.of(1)); var nonZeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); var expectedDelay = Duration.ofMillis(4711); when(retryUtil.calculateDelay(anyInt())).thenReturn(expectedDelay); terminationHandler.transitionInto(nonZeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.retryAfter(WORKFLOW_INSTANCE, expectedDelay.toMillis()), COUNTER); }
@Test public void shouldScheduleRetryOnRetryConditionMet() { var data = data(1, 1.0, Optional.of(1)); var nonZeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); Duration expectedDelay = Duration.ofMillis(4711); when(retryUtil.calculateDelay(anyInt())).thenReturn(expectedDelay); when(workflows.get()).thenReturn(Map.of(WORKFLOW_WITH_RETRY_CONDITION.id(), WORKFLOW_WITH_RETRY_CONDITION)); terminationHandler.transitionInto(nonZeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.retryAfter(WORKFLOW_INSTANCE, expectedDelay.toMillis()), COUNTER); }
@Test public void shouldStopOnRetryConditionNotMet() { var data = data(1, 1.0, Optional.of(2)); var nonZeroTerm = RunState.create(WORKFLOW_INSTANCE, TERMINATED, data, NOW, COUNTER); when(workflows.get()).thenReturn(Map.of(WORKFLOW_WITH_RETRY_CONDITION.id(), WORKFLOW_WITH_RETRY_CONDITION)); terminationHandler.transitionInto(nonZeroTerm, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.stop(WORKFLOW_INSTANCE), COUNTER); }