ExecutionFlowDao { public void uploadExecutableFlow(final ExecutableFlow flow) throws ExecutorManagerException { final String useExecutorParam = flow.getExecutionOptions().getFlowParameters().get(ExecutionOptions.USE_EXECUTOR); final String executorId = StringUtils.isNotEmpty(useExecutorParam) ? useExecutorParam : null; final String flowPriorityParam = flow.getExecutionOptions().getFlowParameters().get(ExecutionOptions.FLOW_PRIORITY); final int flowPriority = StringUtils.isNotEmpty(flowPriorityParam) ? Integer.parseInt(flowPriorityParam) : ExecutionOptions.DEFAULT_FLOW_PRIORITY; final String INSERT_EXECUTABLE_FLOW = "INSERT INTO execution_flows " + "(project_id, flow_id, version, status, submit_time, submit_user, update_time, " + "use_executor, flow_priority, execution_source) values (?,?,?,?,?,?,?,?,?,?)"; final long submitTime = flow.getSubmitTime(); final String executionSource = flow.getExecutionSource(); final SQLTransaction<Long> insertAndGetLastID = transOperator -> { transOperator.update(INSERT_EXECUTABLE_FLOW, flow.getProjectId(), flow.getFlowId(), flow.getVersion(), flow.getStatus().getNumVal(), submitTime, flow.getSubmitUser(), submitTime, executorId, flowPriority, executionSource); transOperator.getConnection().commit(); return transOperator.getLastInsertId(); }; try { final long id = this.dbOperator.transaction(insertAndGetLastID); logger.info("Flow given " + flow.getFlowId() + " given id " + id); flow.setExecutionId((int) id); updateExecutableFlow(flow); } catch (final SQLException e) { throw new ExecutorManagerException("Error creating execution.", 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 testAssignAndUnassignExecutor() throws Exception { final String host = "localhost"; final int port = 12345; final Executor executor = this.executorDao.addExecutor(host, port); final ExecutableFlow flow = TestUtils.createTestExecutableFlow("exectest1", "exec1"); this.executionFlowDao.uploadExecutableFlow(flow); this.assignExecutor.assignExecutor(executor.getId(), flow.getExecutionId()); final Executor fetchExecutor = this.executorDao .fetchExecutorByExecutionId(flow.getExecutionId()); assertThat(fetchExecutor).isEqualTo(executor); this.assignExecutor.unassignExecutor(flow.getExecutionId()); assertThat(this.executorDao.fetchExecutorByExecutionId(flow.getExecutionId())).isNull(); }
@Test public void testAssignExecutorInvalidExecutor() throws Exception { final ExecutableFlow flow = TestUtils.createTestExecutableFlow("exectest1", "exec1"); this.executionFlowDao.uploadExecutableFlow(flow); assertThatThrownBy( () -> this.assignExecutor.assignExecutor(1, flow.getExecutionId())) .isInstanceOf(ExecutorManagerException.class) .hasMessageContaining("non-existent executor"); }