WorkflowDispatcher implements Runnable { @Override public void run() { logger.info("Dispacther started."); try { workflowDefinitions.postProcessWorkflowDefinitions(); running = true; while (!shutdownRequested) { if (paused) { sleep(false); } else { try { executor.waitUntilQueueSizeLowerThanThreshold(executorDao.getMaxWaitUntil()); if (!shutdownRequested) { if (executorDao.tick()) { workflowInstances.recoverWorkflowInstancesFromDeadNodes(); } int potentiallyStuckProcessors = stateProcessorFactory.getPotentiallyStuckProcessors(); if (potentiallyStuckProcessors > 0) { periodicLogger.warn("{} of {} state processor threads are potentially stuck (processing longer than {} seconds)", potentiallyStuckProcessors, executor.getThreadCount(), stuckThreadThresholdSeconds); } dispatch(getNextInstanceIds()); } } catch (PollingRaceConditionException pex) { logger.debug(pex.getMessage()); sleep(true); } catch (PollingBatchException pex) { logger.warn(pex.getMessage()); } catch (@SuppressWarnings("unused") InterruptedException dropThrough) { } catch (Exception e) { logger.error("Exception in executing dispatcher - retrying after sleep period ({})", e.getMessage(), e); sleep(false); } } } } finally { shutdownPool(); executorDao.markShutdown(); running = false; logger.info("Shutdown completed."); shutdownDone.countDown(); } } @Inject @SuppressFBWarnings(value = "WEM_WEAK_EXCEPTION_MESSAGING", justification = "Transaction support exception message is fine")  WorkflowDispatcher(WorkflowInstanceExecutor executor, WorkflowInstanceDao workflowInstances,
      WorkflowStateProcessorFactory stateProcessorFactory, WorkflowDefinitionService workflowDefinitions, ExecutorDao executorDao,
      Environment env); @Override void run(); void shutdown(); void pause(); void resume(); boolean isPaused(); boolean isRunning();  }
@Test public void errorDuringDispatcherExecutionStopsDispatcher() throws Throwable { @SuppressWarnings("unused") class ErrorDuringDispatcherExecutionStopsDispatcher extends MultithreadedTestCase { public void threadDispatcher() { when(workflowInstances.pollNextWorkflowInstanceIds(anyInt())).thenThrow(new AssertionError()).thenReturn(ids(1L)); try { dispatcher.run(); Assertions.fail("Error should stop the dispatcher"); } catch (AssertionError expected) { assertPoolIsShutdown(true); } } @Override public void finish() { verify(workflowInstances).pollNextWorkflowInstanceIds(anyInt()); verify(executorFactory, never()).createProcessor(anyLong(), any()); } } runOnce(new ErrorDuringDispatcherExecutionStopsDispatcher()); }