ExecutionController extends EventHandler implements ExecutorManagerAdapter { @Override public void cancelFlow(final ExecutableFlow exFlow, final String userId) throws ExecutorManagerException { synchronized (exFlow) { final Map<Integer, Pair<ExecutionReference, ExecutableFlow>> unfinishedFlows = this.executorLoader .fetchUnfinishedFlows(); if (unfinishedFlows.containsKey(exFlow.getExecutionId())) { final Pair<ExecutionReference, ExecutableFlow> pair = unfinishedFlows .get(exFlow.getExecutionId()); if (pair.getFirst().getExecutor().isPresent()) { this.apiGateway .callWithReferenceByUser(pair.getFirst(), ConnectorParams.CANCEL_ACTION, userId); } else { ExecutionControllerUtils.finalizeFlow(this.executorLoader, this.alerterHolder, exFlow, "Cancelled before dispatching to executor", null); } } else { throw new ExecutorManagerException("Execution " + exFlow.getExecutionId() + " of flow " + exFlow.getFlowId() + " isn't running."); } } } @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 testKillQueuedFlow() throws Exception { submitFlow(this.flow1, this.ref1); this.flow1.setStatus(Status.PREPARING); this.controller.cancelFlow(this.flow1, this.user.getUserId()); assertThat(this.flow1.getStatus()).isEqualTo(Status.FAILED); this.flow1.getExecutableNodes().forEach(node -> { assertThat(node.getStatus()).isEqualTo(Status.KILLING); }); }
@Test public void testKillRunningFlow() throws Exception { submitFlow(this.flow2, this.ref2); this.flow2.setStatus(Status.RUNNING); this.controller.cancelFlow(this.flow2, this.user.getUserId()); verify(this.apiGateway).callWithReferenceByUser(this.ref2, ConnectorParams.CANCEL_ACTION, this.user.getUserId()); }