WorkflowInstanceDao { public long insertWorkflowInstance(WorkflowInstance instance) { long id; if (sqlVariants.hasUpdateableCTE()) { id = insertWorkflowInstanceWithCte(instance); } else { id = insertWorkflowInstanceWithTransaction(instance); } if (instance.nextActivation != null && instance.nextActivation.isBeforeNow()) { workflowInstanceExecutor.wakeUpDispatcherIfNeeded(); } return id; } @Inject  WorkflowInstanceDao(SQLVariants sqlVariants, @NFlow JdbcTemplate nflowJdbcTemplate,
      @NFlow TransactionTemplate transactionTemplate, @NFlow NamedParameterJdbcTemplate nflowNamedParameterJdbcTemplate,
      ExecutorDao executorDao, WorkflowInstanceExecutor workflowInstanceExecutor, WorkflowInstanceFactory workflowInstanceFactory,
      Environment env); long insertWorkflowInstance(WorkflowInstance instance); @SuppressWarnings("null") void updateWorkflowInstanceAfterExecution(WorkflowInstance instance, WorkflowInstanceAction action,
      List<WorkflowInstance> childWorkflows, List<WorkflowInstance> workflows, boolean createAction); int updateWorkflowInstance(WorkflowInstance instance); void recoverWorkflowInstancesFromDeadNodes(); void checkStateVariableValueLength(String name, String value); boolean updateNotRunningWorkflowInstance(WorkflowInstance instance); boolean wakeUpWorkflowExternally(long workflowInstanceId, List<String> expectedStates); boolean wakeupWorkflowInstanceIfNotExecuting(long workflowInstanceId, List<String> expectedStates); WorkflowInstance getWorkflowInstance(long id, Set<WorkflowInstanceInclude> includes, Long maxActions); List<Long> pollNextWorkflowInstanceIds(final int batchSize); List<WorkflowInstance> queryWorkflowInstances(QueryWorkflowInstances query); Stream<WorkflowInstance> queryWorkflowInstancesAsStream(QueryWorkflowInstances query); @Transactional(propagation = MANDATORY) long insertWorkflowInstanceAction(final WorkflowInstance instance, final WorkflowInstanceAction action); long insertWorkflowInstanceAction(final WorkflowInstanceAction action); String getWorkflowInstanceState(long workflowInstanceId); Optional<Integer> getSignal(long workflowInstanceId); @Transactional boolean setSignal(long workflowInstanceId, Optional<Integer> signal, String reason, WorkflowActionType actionType); String getWorkflowInstanceType(long workflowInstanceId);  }
@Test public void pollNextWorkflowInstancesWithPartialRaceCondition() throws InterruptedException { int batchSize = 100; for (int i = 0; i < batchSize; i++) { WorkflowInstance instance = constructWorkflowInstanceBuilder().setNextActivation(now().minusMinutes(1)) .setExecutorGroup("junit").build(); dao.insertWorkflowInstance(instance); } Poller[] pollers = new Poller[] { new Poller(dao, batchSize), new Poller(dao, batchSize) }; for (int i = 0; i < 10; ++i) { Thread[] threads = new Thread[] { new Thread(pollers[0]), new Thread(pollers[1]) }; threads[0].start(); threads[1].start(); threads[0].join(); threads[1].join(); assertThat(pollers[0].returnSize + pollers[1].returnSize, is(batchSize)); if (pollers[0].detectedRaceCondition || pollers[1].detectedRaceCondition || (pollers[0].returnSize < batchSize && pollers[1].returnSize < batchSize)) { return; } } fail("Race condition should happen"); }