ExecutionFlowDao { public int selectAndUpdateExecutionWithLocking(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 -> { final String POLLING_LOCK_NAME = "execution_flows_polling"; final int GET_LOCK_TIMEOUT_IN_SECONDS = 5; int execId = -1; final boolean hasLocked = this.mysqlNamedLock.getLock(transOperator, POLLING_LOCK_NAME, GET_LOCK_TIMEOUT_IN_SECONDS); logger.info("ExecutionFlow polling lock value: " + hasLocked + " for executorId: " + executorId); if (hasLocked) { try { final List<Integer> execIds = transOperator.query(selectExecutionForUpdate, new SelectFromExecutionFlows(), executorId); if (CollectionUtils.isNotEmpty(execIds)) { execId = execIds.get(0); transOperator.update(UPDATE_EXECUTION, executorId, System.currentTimeMillis(), execId); } } finally { this.mysqlNamedLock.releaseLock(transOperator, POLLING_LOCK_NAME); logger.info("Released polling lock for executorId: " + executorId); } } else { logger.info("Could not acquire polling lock for executorId: " + executorId); } 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 testLockSuccessSelectAndUpdateExecutionWithLocking() throws Exception { when(mysqlNamedLock.getLock(any(DatabaseTransOperator.class), any(String.class), any(Integer.class))) .thenReturn(true); when(mysqlNamedLock.releaseLock(any(DatabaseTransOperator.class), any(String.class))).thenReturn(true); final long currentTime = System.currentTimeMillis(); final ExecutableFlow flow1 = submitNewFlow("exectest1", "exec1", currentTime, ExecutionOptions.DEFAULT_FLOW_PRIORITY); final Executor executor1 = this.executorDao.addExecutor("localhost", 12345); assertThat(this.executionFlowDao.selectAndUpdateExecutionWithLocking(executor1.getId(), true)) .isEqualTo(flow1.getExecutionId()); }
@Test public void testLockFailureSelectAndUpdateExecutionWithLocking() throws Exception { when(mysqlNamedLock.getLock(any(DatabaseTransOperator.class), any(String.class), any(Integer.class))) .thenReturn(false); final long currentTime = System.currentTimeMillis(); final ExecutableFlow flow1 = submitNewFlow("exectest1", "exec1", currentTime, ExecutionOptions.DEFAULT_FLOW_PRIORITY); final Executor executor1 = this.executorDao.addExecutor("localhost", 12345); assertThat(this.executionFlowDao.selectAndUpdateExecutionWithLocking(executor1.getId(), true)).isEqualTo(-1); }