PersistentStateManager implements StateManager { @Override public void trigger(WorkflowInstance workflowInstance, Trigger trigger, TriggerParameters parameters) throws IsClosedException { ensureRunning(); log.debug("Trigger {}", workflowInstance); var counter = latestEventCounter(workflowInstance); var initialRunState = RunState.create(workflowInstance, State.NEW, time.get(), counter); var event = Event.triggerExecution(workflowInstance, trigger, parameters); var runState = initialRunState.transition(event, time); try { storage.runInTransactionWithRetries(tx -> { final Optional<Workflow> workflow = tx.workflow(workflowInstance.workflowId()); if (workflow.isEmpty()) { throw new IllegalArgumentException( "Workflow not found: " + workflowInstance.workflowId().toKey()); } return tx.writeActiveState(workflowInstance, runState); }); } catch (TransactionException e) { if (e.isAlreadyExists()) { throw new AlreadyInitializedException("Workflow instance is already triggered: " + workflowInstance); } else if (e.isConflict()) { log.debug("Transaction conflict when triggering workflow instance. Aborted: {}", workflowInstance); throw new RuntimeException(e); } else { log.debug("Transaction failure when triggering workflow instance: {}: {}", workflowInstance, e.getMessage(), e); throw new RuntimeException(e); } } catch (Exception e) { log.debug("Failure when triggering workflow instance: {}: {}", workflowInstance, e.getMessage(), e); Throwables.throwIfUnchecked(e); throw new RuntimeException(e); } postTransition(event, runState); } 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 shouldReInitializeWFInstanceFromNextCounter() throws Exception { when(storage.getLatestStoredCounter(INSTANCE)).thenReturn(Optional.of(INSTANCE_NEW_STATE.counter())); when(transaction.workflow(INSTANCE.workflowId())).thenReturn(Optional.of(WORKFLOW)); stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); verify(storage).getLatestStoredCounter(INSTANCE); verify(transaction).writeActiveState(INSTANCE, INSTANCE_QUEUED_STATE); verify(storage).writeEvent(SequenceEvent.create( Event.triggerExecution(INSTANCE, TRIGGER1, PARAMETERS), INSTANCE_QUEUED_STATE.counter(), NOW.toEpochMilli())); }
@Test public void shouldFailTriggerWFIfAlreadyActive() throws Exception { reset(storage); when(storage.getLatestStoredCounter(any())).thenReturn(Optional.empty()); when(transaction.workflow(INSTANCE.workflowId())).thenReturn(Optional.of(WORKFLOW)); DatastoreException datastoreException = new DatastoreException(1, "", ""); TransactionException transactionException = spy(new TransactionException(datastoreException)); when(transactionException.isAlreadyExists()).thenReturn(true); doThrow(transactionException).when(transaction).writeActiveState(any(), any()); when(storage.runInTransactionWithRetries(any())).thenAnswer(a -> a.<TransactionFunction>getArgument(0).apply(transaction)); try { stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); fail(); } catch (AlreadyInitializedException ignore) { } }
@Test public void shouldFailTriggerIfGetLatestCounterFails() throws Exception { var cause = new IOException(); when(storage.getLatestStoredCounter(any())).thenThrow(cause); exception.expectCause(is(cause)); stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); }
@Test public void shouldFailTriggerIfWorkflowNotFound() throws Exception { when(storage.getLatestStoredCounter(any())).thenReturn(Optional.empty()); when(transaction.workflow(INSTANCE.workflowId())).thenReturn(Optional.empty()); exception.expect(instanceOf(IllegalArgumentException.class)); exception.expectMessage("Workflow not found: " + INSTANCE.workflowId().toKey()); stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); }
@Test public void shouldFailTriggerOnExceptionFromTransaction() throws Exception { reset(storage); when(storage.getLatestStoredCounter(any())).thenReturn(Optional.empty()); var cause = new IOException("fail!"); when(storage.runInTransactionWithRetries(any())).thenThrow(cause); exception.expectCause(is(cause)); stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); }
@Test public void triggerShouldHandleThrowingOutputHandler() throws Exception { when(storage.getLatestStoredCounter(any())).thenReturn(Optional.of(-1L)); when(transaction.workflow(INSTANCE.workflowId())).thenReturn(Optional.of(WORKFLOW)); final RuntimeException rootCause = new RuntimeException("foo!"); doThrow(rootCause).when(outputHandler).transitionInto(any(), any()); try { stateManager.trigger(INSTANCE, TRIGGER1, PARAMETERS); fail(); } catch (Exception e) { assertThat(Throwables.getRootCause(e), is(rootCause)); } }
@Test public void triggerShouldLogTransactionConflict() throws Exception { final TransactionException cause = new TransactionException( new DatastoreException(10, "foo", "bar")); reset(storage); when(storage.getLatestStoredCounter(INSTANCE)).thenReturn(Optional.empty()); when(storage.runInTransactionWithRetries(any())).thenThrow(cause); try { stateManager.trigger(INSTANCE, Trigger.natural(), PARAMETERS); fail(); } catch (Exception ignore) { } verify(logger).debug("Transaction conflict when triggering workflow instance. Aborted: {}", INSTANCE); }
@Test public void triggerShouldLogTransactionFailure() throws Exception { final TransactionException cause = new TransactionException( new DatastoreException(new IOException("netsplit!"))); reset(storage); when(storage.getLatestStoredCounter(INSTANCE)).thenReturn(Optional.empty()); when(storage.runInTransactionWithRetries(any())).thenThrow(cause); try { stateManager.trigger(INSTANCE, Trigger.natural(), PARAMETERS); fail(); } catch (Exception ignore) { } verify(logger).debug("Transaction failure when triggering workflow instance: {}: {}", INSTANCE, cause.getMessage(), cause); }
@Test public void triggerShouldLogFailure() throws Exception { final Exception cause = new Exception("fubared"); reset(storage); when(storage.getLatestStoredCounter(INSTANCE)).thenReturn(Optional.empty()); when(storage.runInTransactionWithRetries(any())).thenThrow(cause); try { stateManager.trigger(INSTANCE, Trigger.natural(), PARAMETERS); fail(); } catch (Exception ignore) { } verify(logger).debug("Failure when triggering workflow instance: {}: {}", INSTANCE, cause.getMessage(), cause); }