PersistentStateManager implements StateManager { @Override public void receive(Event event) throws IsClosedException { receive(event, Long.MAX_VALUE); } PersistentStateManager(
      Time time,
      ExecutorService executor,
      Storage storage,
      BiConsumer<SequenceEvent, RunState> eventConsumer,
      Executor eventConsumerExecutor,
      OutputHandler outputHandler,
      ShardedCounter shardedCounter);   PersistentStateManager(
      Time time,
      ExecutorService executor,
      Storage storage,
      BiConsumer<SequenceEvent, RunState> eventConsumer,
      Executor eventConsumerExecutor,
      OutputHandler outputHandler,
      ShardedCounter shardedCounter,
      Logger logger); @Override void tick(); @Override void trigger(WorkflowInstance workflowInstance, Trigger trigger, TriggerParameters parameters); @Override void receive(Event event); @Override void receive(Event event, long expectedCounter); @Override Set<WorkflowInstance> listActiveInstances(); @Override Map<WorkflowInstance, RunState> getActiveStates(); @Override Map<WorkflowInstance, RunState> getActiveStatesByTriggerId(String triggerId); @Override Optional<RunState> getActiveState(WorkflowInstance workflowInstance); @Override void close();  }
@Test public void shouldNotBeActiveAfterHalt() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.PREPARE, StateData.newBuilder().resourceIds(ImmutableSet.of()).build(), NOW, 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); Event event = Event.halt(INSTANCE); stateManager.receive(event); verify(transaction).deleteActiveState(INSTANCE); verify(storage).writeEvent(SequenceEvent.create(event, 18, NOW.toEpochMilli())); }
@Test public void shouldNotFailWhenMissingResourceIdsWhenTransitionFromPrepareToError() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.PREPARE, StateData.zero(), NOW, 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); Event event = Event.halt(INSTANCE); stateManager.receive(event); verify(transaction).deleteActiveState(INSTANCE); verify(storage).writeEvent(SequenceEvent.create(event, 18, NOW.toEpochMilli())); }
@Test public void shouldWriteEvents() throws Exception { Event event = Event.started(INSTANCE); Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.SUBMITTED, StateData.zero(), NOW, 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); stateManager.receive(event); verify(storage).writeEvent(SequenceEvent.create(event, 18, NOW.toEpochMilli())); }
@Test public void shouldFailReceiveEventWithHigherCounter() throws Exception { Event event = Event.started(INSTANCE); when(transaction.readActiveState(INSTANCE)).thenReturn( Optional.of(RunState.create(INSTANCE, State.SUBMITTED, StateData.zero(), NOW, 17))); try { stateManager.receive(event, 16); fail(); } catch (StateTransitionConflictException ignore) { } verify(storage, never()).writeEvent(any()); }
@Test public void shouldFailReceiveEventWithLowerCounter() throws Exception { Event event = Event.started(INSTANCE); Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.SUBMITTED, StateData.zero(), NOW.minusMillis(1), 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); try { stateManager.receive(event, 18); fail(); } catch (RuntimeException e) { assertThat(e.getMessage(), startsWith("Unexpected current counter is less than last observed one for")); } verify(storage, never()).writeEvent(any()); }
@Test public void shouldRemoveStateIfTerminal() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.TERMINATED, StateData.zero(), NOW, 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); Event event = Event.success(INSTANCE); stateManager.receive(event); verify(transaction).deleteActiveState(INSTANCE); verify(storage).writeEvent(SequenceEvent.create(event, 18, NOW.toEpochMilli())); }
@Test public void shouldWriteActiveStateOnEvent() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.QUEUED, StateData.zero(), NOW, 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); stateManager.receive(Event.dequeue(INSTANCE, ImmutableSet.of())); verify(transaction).updateActiveState(INSTANCE, RunState.create(INSTANCE, State.PREPARE, StateData.newBuilder().resourceIds(ImmutableSet.of()).build(), NOW, 18)); }
@Test public void shouldPreventIllegalStateTransition() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.QUEUED, StateData.zero(), NOW.minusMillis(1), 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); try { stateManager.receive(Event.terminate(INSTANCE, Optional.empty())); fail(); } catch (IllegalStateException ignore) { } verify(transaction, never()).updateActiveState(any(), any()); }
@Test public void shouldFailReceiveForUnknownActiveWFInstance() throws Exception { when(transaction.readActiveState(INSTANCE)).thenReturn(Optional.empty()); try { stateManager.receive(Event.terminate(INSTANCE, Optional.empty())); fail(); } catch (IllegalArgumentException ignore) { } verify(transaction, never()).updateActiveState(any(), any()); }
@Test public void receiveShouldPropagateOutputHandlerException() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.QUEUED, StateData.zero(), NOW.minusMillis(1), 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); final RuntimeException rootCause = new RuntimeException("foo!"); doThrow(rootCause).when(outputHandler).transitionInto(any(), any()); try { stateManager.receive(Event.dequeue(INSTANCE, ImmutableSet.of())); fail(); } catch (Exception e) { assertThat(Throwables.getRootCause(e), is(rootCause)); } }
@Test public void receiveShouldHandleOutputHandlerThrowingStateTransitionConflict() throws Exception { Optional<RunState> runState = Optional.of( RunState.create(INSTANCE, State.QUEUED, StateData.zero(), NOW.minusMillis(1), 17)); when(transaction.readActiveState(INSTANCE)).thenReturn(runState); final StateTransitionConflictException rootCause = new StateTransitionConflictException("conflict!"); doThrow(rootCause).when(outputHandler).transitionInto(any(), any()); stateManager.receive(Event.dequeue(INSTANCE, ImmutableSet.of())); verify(logger).debug("State transition conflict when invoking output handler: {}", INSTANCE, rootCause); }
@Test public void receiveShouldLogFailure() throws Exception { final Exception cause = new Exception("fubared"); reset(storage); when(storage.runInTransactionWithRetries(any())).thenThrow(cause); final Event event = Event.started(INSTANCE); try { stateManager.receive(event, 4711L); fail(); } catch (Exception ignore) { } verify(logger).debug( "Failed workflow instance transition: {}, counter={}", event, 4711L, cause); }
@Test public void shouldThrowRuntimeException() throws Exception { final IOException exception = new IOException(); doThrow(exception).when(storage).runInTransactionWithRetries(any()); try { stateManager.receive(Event.dequeue(INSTANCE, ImmutableSet.of())); fail(); } catch (Exception e) { assertThat(Throwables.getRootCause(e), is(exception)); } }
@Test public void shouldFailToUpdateResourceCountersOnDequeueDueToCapacity() throws Exception { givenState(INSTANCE, State.QUEUED); doThrow(new CounterCapacityException("foo")) .when(transaction).updateCounter(shardedCounter, "resource1", 1); final Set<Resource> resources = ImmutableSet.of(Resource.create("resource1", 1)); final List<String> resourceIds = resources.stream().map(Resource::id).sorted().collect(toList()); final Event dequeueEvent = Event.dequeue(INSTANCE, ImmutableSet.copyOf(resourceIds)); final Event infoEvent = Event.info(INSTANCE, Message.info(String.format("Resource limit reached for: %s", resourceIds))); final PersistentStateManager spied = spy(stateManager); doNothing().when(spied).receiveIgnoreClosed(eq(infoEvent), anyLong()); try { spied.receive(dequeueEvent); fail(); } catch (Exception e) { } verify(spied).receiveIgnoreClosed(eq(infoEvent), anyLong()); }
@Test public void shouldFailToUpdateResourceCountersOnDequeueDueToCapacityAndNoInfoEventSent() throws Exception { final Set<Resource> resources = ImmutableSet.of(Resource.create("resource1", 1)); final List<String> resourceIds = resources.stream().map(Resource::id).sorted().collect(toList()); final Message message = Message.info(String.format("Resource limit reached for: %s", resourceIds)); final RunState runState = RunState.create(INSTANCE, State.QUEUED, STATE_DATA_1.builder().messages(message).build(), NOW.minusMillis(1), 17); when(transaction.readActiveState(INSTANCE)).thenReturn(Optional.of(runState)); doThrow(new CounterCapacityException("foo")) .when(transaction).updateCounter(shardedCounter, "resource1", 1); final Event dequeueEvent = Event.dequeue(INSTANCE, resources.stream().map(Resource::id).collect(toSet())); final PersistentStateManager spied = spy(stateManager); try { spied.receive(dequeueEvent); fail(); } catch (Exception e) { } verify(spied, never()).receiveIgnoreClosed(eq(Event.info(INSTANCE, message)), anyLong()); }
@Test public void shouldFailToUpdateResourceCountersOnDequeueDueToConflict() throws Exception { givenState(INSTANCE, State.QUEUED); var rootCause = new DatastoreIOException(new DatastoreException(10, "conflict!", "conflict!")); doThrow(rootCause).when(transaction).updateCounter(shardedCounter, "resource1", 1); var dequeueEvent = Event.dequeue(INSTANCE, Set.of("resource1")); exception.expect(RuntimeException.class); exception.expectCause(is(rootCause)); stateManager.receive(dequeueEvent); }