ExecutorManager 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 = ""; if (this.queuedFlows.isFull()) { message = String .format( "Failed to submit %s for project %s. Azkaban has overrun its webserver queue capacity", flowId, exflow.getProjectName()); logger.error(message); this.commonMetrics.markSubmitFlowFail(); } else { 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); final ExecutionReference reference = new ExecutionReference(exflow.getExecutionId()); this.executorLoader.addActiveExecutableReference(reference); this.queuedFlows.enqueue(exflow, reference); message += "Execution queued successfully with exec id " + exflow.getExecutionId(); this.commonMetrics.markSubmitFlowSuccess(); } return message; } } @Inject  ExecutorManager(final Props azkProps, final ExecutorLoader executorLoader,
      final CommonMetrics commonMetrics,
      final ExecutorApiGateway apiGateway,
      final RunningExecutions runningExecutions,
      final ActiveExecutors activeExecutors,
      final ExecutorManagerUpdaterStage updaterStage,
      final ExecutionFinalizer executionFinalizer,
      final RunningExecutionsUpdaterThread updaterThread); @Override void start(); @Override void setupExecutors(); @Override void disableQueueProcessorThread(); @Override void enableQueueProcessorThread(); State getQueueProcessorThreadState(); boolean isQueueProcessorThreadActive(); long getLastSuccessfulExecutorInfoRefresh(); Set<String> getAvailableExecutorComparatorNames(); Set<String> getAvailableExecutorFilterNames(); @Override State getExecutorManagerThreadState(); String getExecutorThreadStage(); @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(); String getRunningFlowIds(); String 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 String submitExecutableFlow(final ExecutableFlow exflow, final String userId); @Override Map<String, String> doRampActions(final List<Map<String, Object>> rampActions); @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 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(expected = ExecutorManagerException.class) public void testDuplicateQueuedFlows() throws Exception { final ExecutorManager manager = createMultiExecutorManagerInstance(); final ExecutableFlow flow1 = TestUtils.createTestExecutableFlow("exectest1", "exec1"); flow1.getExecutionOptions().setConcurrentOption( ExecutionOptions.CONCURRENT_OPTION_SKIP); final User testUser = TestUtils.getTestUser(); manager.submitExecutableFlow(flow1, testUser.getUserId()); manager.submitExecutableFlow(flow1, testUser.getUserId()); }
@Test public void testSubmitFlows() throws Exception { testSetUpForRunningFlows(); final ExecutableFlow flow1 = TestUtils.createTestExecutableFlow("exectest1", "exec1"); this.manager.submitExecutableFlow(flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow1); verify(this.loader).addActiveExecutableReference(any()); }
@Test(expected = ExecutorManagerException.class) public void testTooManySubmitFlows() throws Exception { testSetUpForRunningFlows(); final ExecutableFlow flow1 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow1.setExecutionId(101); final ExecutableFlow flow2 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow2.setExecutionId(102); final ExecutableFlow flow3 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow3.setExecutionId(103); final ExecutableFlow flow4 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow4.setExecutionId(104); this.manager.submitExecutableFlow(flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow1); this.manager.submitExecutableFlow(flow2, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow2); this.manager.submitExecutableFlow(flow3, this.user.getUserId()); this.manager.submitExecutableFlow(flow4, this.user.getUserId()); }
@Test(expected = ExecutorManagerException.class) public void testConcurrentRunWhitelist() throws Exception { testSetUpForRunningFlows(); this.props.put(ConfigurationKeys.CONCURRENT_RUNS_ONEFLOW_WHITELIST, "basicyamlshelltest," + "bashSleep,4"); final ExecutableFlow flow1 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow1.setExecutionId(101); final ExecutableFlow flow2 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow2.setExecutionId(102); final ExecutableFlow flow3 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow3.setExecutionId(103); final ExecutableFlow flow4 = TestUtils .createTestExecutableFlowFromYaml("basicyamlshelltest", "bashSleep"); flow4.setExecutionId(104); this.manager.submitExecutableFlow(flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow1); this.manager.submitExecutableFlow(flow2, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow2); this.manager.submitExecutableFlow(flow3, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow3); this.manager.submitExecutableFlow(flow4, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow4); }
@Test public void testSetFlowLock() throws Exception { testSetUpForRunningFlows(); final ExecutableFlow flow1 = TestUtils.createTestExecutableFlow("exectest1", "exec1"); flow1.setLocked(true); final String msg = this.manager.submitExecutableFlow(flow1, this.user.getUserId()); assertThat(msg).isEqualTo("Flow derived-member-data for project flow is locked."); flow1.setLocked(false); this.manager.submitExecutableFlow(flow1, this.user.getUserId()); verify(this.loader).uploadExecutableFlow(flow1); verify(this.loader).addActiveExecutableReference(any()); }