ExecutionController extends EventHandler implements ExecutorManagerAdapter { @Override public String submitExecutableFlow(final ExecutableFlow exflow, final String userId) throws ExecutorManagerException { if (exflow.isLocked()) { final String message = String.format("Flow %s for project %s is locked.", exflow.getId(), exflow.getProjectName()); logger.info(message); return message; } final String exFlowKey = exflow.getProjectName() + "." + exflow.getId() + ".submitFlow"; synchronized (exFlowKey.intern()) { final String flowId = exflow.getFlowId(); logger.info("Submitting execution flow " + flowId + " by " + userId); String message = ""; final int projectId = exflow.getProjectId(); exflow.setSubmitUser(userId); exflow.setStatus(Status.PREPARING); exflow.setSubmitTime(System.currentTimeMillis()); final List<Integer> running = getRunningFlows(projectId, flowId); ExecutionOptions options = exflow.getExecutionOptions(); if (options == null) { options = new ExecutionOptions(); } if (options.getDisabledJobs() != null) { FlowUtils.applyDisabledJobs(options.getDisabledJobs(), exflow); } if (!running.isEmpty()) { final int maxConcurrentRuns = ExecutorUtils.getMaxConcurrentRunsForFlow( exflow.getProjectName(), flowId, this.maxConcurrentRunsOneFlow, this.maxConcurrentRunsPerFlowMap); if (running.size() > maxConcurrentRuns) { this.commonMetrics.markSubmitFlowSkip(); throw new ExecutorManagerException("Flow " + flowId + " has more than " + maxConcurrentRuns + " concurrent runs. Skipping", ExecutorManagerException.Reason.SkippedExecution); } else if (options.getConcurrentOption().equals( ExecutionOptions.CONCURRENT_OPTION_PIPELINE)) { Collections.sort(running); final Integer runningExecId = running.get(running.size() - 1); options.setPipelineExecutionId(runningExecId); message = "Flow " + flowId + " is already running with exec id " + runningExecId + ". Pipelining level " + options.getPipelineLevel() + ". \n"; } else if (options.getConcurrentOption().equals( ExecutionOptions.CONCURRENT_OPTION_SKIP)) { this.commonMetrics.markSubmitFlowSkip(); throw new ExecutorManagerException("Flow " + flowId + " is already running. Skipping execution.", ExecutorManagerException.Reason.SkippedExecution); } else { message = "Flow " + flowId + " is already running with exec id " + StringUtils.join(running, ",") + ". Will execute concurrently. \n"; } } final boolean memoryCheck = !ProjectWhitelist.isProjectWhitelisted(exflow.getProjectId(), ProjectWhitelist.WhitelistType.MemoryCheck); options.setMemoryCheck(memoryCheck); this.executorLoader.uploadExecutableFlow(exflow); this.commonMetrics.markSubmitFlowSuccess(); message += "Execution queued successfully with exec id " + exflow.getExecutionId(); return message; } } @Inject  ExecutionController(final Props azkProps, final ExecutorLoader executorLoader,
      final CommonMetrics commonMetrics,
      final ExecutorApiGateway apiGateway, final AlerterHolder alerterHolder, final
  ExecutorHealthChecker executorHealthChecker); @Override void setupExecutors(); @Override void disableQueueProcessorThread(); @Override void enableQueueProcessorThread(); @Override State getExecutorManagerThreadState(); @Override boolean isExecutorManagerThreadActive(); @Override long getLastExecutorManagerThreadCheckTime(); @Override Collection<Executor> getAllActiveExecutors(); @Override Executor fetchExecutor(final int executorId); @Override Set<String> getPrimaryServerHosts(); @Override Set<String> getAllActiveExecutorServerHosts(); @Override List<Integer> getRunningFlows(final int projectId, final String flowId); @Override List<Pair<ExecutableFlow, Optional<Executor>>> getActiveFlowsWithExecutor(); @Override boolean isFlowRunning(final int projectId, final String flowId); @Override ExecutableFlow getExecutableFlow(final int execId); @Override List<ExecutableFlow> getRunningFlows(); List<Integer> getRunningFlowIds(); List<Integer> getQueuedFlowIds(); @Override long getQueuedFlowSize(); @Override long getAgedQueuedFlowSize(); @Override DispatchMethod getDispatchMethod(); @Override List<ExecutableFlow> getRecentlyFinishedFlows(); @Override List<ExecutableFlow> getExecutableFlows(final int skip, final int size); @Override List<ExecutableFlow> getExecutableFlows(final String flowIdContains,
      final int skip, final int size); @Override List<ExecutableFlow> getExecutableFlows(final String projContain,
      final String flowContain, final String userContain, final int status, final long begin,
      final long end,
      final int skip, final int size); @Override List<ExecutableJobInfo> getExecutableJobs(final Project project,
      final String jobId, final int skip, final int size); @Override int getNumberOfJobExecutions(final Project project, final String jobId); @Override LogData getExecutableFlowLog(final ExecutableFlow exFlow, final int offset,
      final int length); @Override LogData getExecutionJobLog(final ExecutableFlow exFlow, final String jobId,
      final int offset, final int length, final int attempt); @Override List<Object> getExecutionJobStats(final ExecutableFlow exFlow, final String jobId,
      final int attempt); @Override Map<String, String> getExternalJobLogUrls(final ExecutableFlow exFlow, final String jobId,
      final int attempt); @Override void cancelFlow(final ExecutableFlow exFlow, final String userId); @Override void resumeFlow(final ExecutableFlow exFlow, final String userId); @Override void pauseFlow(final ExecutableFlow exFlow, final String userId); @Override void retryFailures(final ExecutableFlow exFlow, final String userId); @Override Map<String, String> doRampActions(List<Map<String, Object>> rampActions); @Override String submitExecutableFlow(final ExecutableFlow exflow, final String userId); @Override Map<String, Object> callExecutorStats(final int executorId, final String action,
      final Pair<String, String>... params); @Override Map<String, Object> callExecutorJMX(final String hostPort, final String action,
      final String mBean); @Override void start(); @Override void shutdown(); @Override int getExecutableFlows(final int projectId, final String flowId, final int from,
      final int length, final List<ExecutableFlow> outputList); @Override List<ExecutableFlow> getExecutableFlows(final int projectId, final String flowId,
      final int from, final int length, final Status status);  }
@Test public void testSubmitFlows() throws Exception { this.controller.submitExecutableFlow(this.flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(this.flow1); }
@Test public void testSubmitFlowsExceedingMaxConcurrentRuns() throws Exception { this.props.put(ConfigurationKeys.CONCURRENT_RUNS_ONEFLOW_WHITELIST, "exectest1," + "exec2,3"); submitFlow(this.flow2, this.ref2); submitFlow(this.flow3, this.ref3); assertThatThrownBy(() -> this.controller.submitExecutableFlow(this.flow4, this.user.getUserId ())).isInstanceOf(ExecutorManagerException.class).hasMessageContaining("Flow " + this .flow4.getId() + " has more than 1 concurrent runs. Skipping"); }
@Test public void testSubmitFlowsConcurrentWhitelist() throws Exception { this.props.put(Constants.ConfigurationKeys.MAX_CONCURRENT_RUNS_ONEFLOW, 1); submitFlow(this.flow2, this.ref2); submitFlow(this.flow3, this.ref3); assertThatThrownBy(() -> this.controller.submitExecutableFlow(this.flow4, this.user.getUserId ())).isInstanceOf(ExecutorManagerException.class).hasMessageContaining("Flow " + this .flow4.getId() + " has more than 1 concurrent runs. Skipping"); }
@Test public void testSubmitFlowsWithSkipOption() throws Exception { submitFlow(this.flow2, this.ref2); this.flow3.getExecutionOptions().setConcurrentOption(ExecutionOptions.CONCURRENT_OPTION_SKIP); assertThatThrownBy( () -> this.controller.submitExecutableFlow(this.flow3, this.user.getUserId())) .isInstanceOf(ExecutorManagerException.class).hasMessageContaining( "Flow " + this.flow3.getId() + " is already running. Skipping execution."); }
@Test public void testSetFlowLock() throws Exception { this.flow1.setLocked(true); final String msg = this.controller.submitExecutableFlow(this.flow1, this.user.getUserId()); assertThat(msg).isEqualTo("Flow derived-member-data for project flow is locked."); this.flow1.setLocked(false); this.controller.submitExecutableFlow(this.flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(this.flow1); }