ExecutorManager extends EventHandler implements
    ExecutorManagerAdapter { @Override public Collection<Executor> getAllActiveExecutors() { return Collections.unmodifiableCollection(this.activeExecutors.getAll()); } @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 public void testMultipleExecutorScenario() throws Exception { this.props.put(Constants.ConfigurationKeys.USE_MULTIPLE_EXECUTORS, "true"); final Executor executor1 = this.loader.addExecutor("localhost", 12345); final Executor executor2 = this.loader.addExecutor("localhost", 12346); final ExecutorManager manager = createExecutorManager(); final Set<Executor> activeExecutors = new HashSet(manager.getAllActiveExecutors()); Assert.assertArrayEquals(activeExecutors.toArray(), new Executor[]{ executor1, executor2}); }