ExecutionFlowDao { public int selectAndUpdateExecution(final int executorId, final boolean isActive) throws ExecutorManagerException { final String UPDATE_EXECUTION = "UPDATE execution_flows SET executor_id = ?, update_time = ? " + "where exec_id = ?"; final String selectExecutionForUpdate = isActive ? SelectFromExecutionFlows.SELECT_EXECUTION_FOR_UPDATE_ACTIVE : SelectFromExecutionFlows.SELECT_EXECUTION_FOR_UPDATE_INACTIVE; final SQLTransaction<Integer> selectAndUpdateExecution = transOperator -> { transOperator.getConnection().setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED); final List<Integer> execIds = transOperator.query(selectExecutionForUpdate, new SelectFromExecutionFlows(), executorId); int execId = -1; if (!execIds.isEmpty()) { execId = execIds.get(0); transOperator.update(UPDATE_EXECUTION, executorId, System.currentTimeMillis(), execId); } transOperator.getConnection().commit(); return execId; }; try { return this.dbOperator.transaction(selectAndUpdateExecution); } catch (final SQLException e) { throw new ExecutorManagerException("Error selecting and updating execution with executor " + executorId, e); } } @Inject  ExecutionFlowDao(final DatabaseOperator dbOperator, final MysqlNamedLock mysqlNamedLock); void uploadExecutableFlow(final ExecutableFlow flow); List<ExecutableFlow> fetchAgedQueuedFlows(final Duration minAge); List<Pair<ExecutionReference, ExecutableFlow>> fetchQueuedFlows(); List<ExecutableFlow> fetchFlowHistory(final int projectId, final String flowId, final
  long startTime); ExecutableFlow fetchExecutableFlow(final int execId); void unsetExecutorIdForExecution(final int executionId); int selectAndUpdateExecution(final int executorId, final boolean isActive); int selectAndUpdateExecutionWithLocking(final int executorId, final boolean isActive);  }
@Test public void testSelectAndUpdateExecution() throws Exception { final ExecutableFlow flow = TestUtils.createTestExecutableFlow("exectest1", "exec1"); flow.setStatus(Status.PREPARING); flow.setSubmitTime(System.currentTimeMillis()); this.executionFlowDao.uploadExecutableFlow(flow); final Executor executor = this.executorDao.addExecutor("localhost", 12345); assertThat(this.executionFlowDao.selectAndUpdateExecution(executor.getId(), true)) .isEqualTo(flow.getExecutionId()); assertThat(this.executorDao.fetchExecutorByExecutionId(flow.getExecutionId())).isEqualTo (executor); }
@Test public void testSelectAndUpdateExecutionWithPriority() throws Exception { assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected no execution selected") .isEqualTo(-1); final long currentTime = System.currentTimeMillis(); final ExecutableFlow lowPriorityFlow1 = submitNewFlow("exectest1", "exec1", currentTime, ExecutionOptions.DEFAULT_FLOW_PRIORITY); final ExecutableFlow highPriorityFlow = submitNewFlow("exectest1", "exec1", currentTime + 5, ExecutionOptions.DEFAULT_FLOW_PRIORITY + 5); final ExecutableFlow lowPriorityFlow2 = submitNewFlow("exectest1", "exec1", currentTime + 10, ExecutionOptions.DEFAULT_FLOW_PRIORITY + 3); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected flow with highest priority") .isEqualTo(highPriorityFlow.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected second flow with highest priority") .isEqualTo(lowPriorityFlow2.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected flow with lowest priority") .isEqualTo(lowPriorityFlow1.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected no execution selected") .isEqualTo(-1); }
@Test public void testSelectAndUpdateExecutionWithSamePriority() throws Exception { assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected no execution selected") .isEqualTo(-1); final long currentTime = System.currentTimeMillis(); final ExecutableFlow submittedFlow1 = submitNewFlow("exectest1", "exec1", currentTime, ExecutionOptions.DEFAULT_FLOW_PRIORITY + 3); final ExecutableFlow submittedFlow2 = submitNewFlow("exectest1", "exec1", currentTime + 5, ExecutionOptions.DEFAULT_FLOW_PRIORITY + 3); final ExecutableFlow submittedFlow3 = submitNewFlow("exectest1", "exec1", currentTime + 10, ExecutionOptions.DEFAULT_FLOW_PRIORITY + 3); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected first flow submitted") .isEqualTo(submittedFlow1.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected second flow submitted") .isEqualTo(submittedFlow2.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected last flow submitted") .isEqualTo(submittedFlow3.getExecutionId()); assertThat(this.executionFlowDao.selectAndUpdateExecution(-1, true)) .as("Expected no execution selected") .isEqualTo(-1); }