ExecutionController extends EventHandler implements ExecutorManagerAdapter { Set<String> getApplicationIds(final ExecutableFlow exFlow, final String jobId, final int attempt) { final Set<String> applicationIds = new LinkedHashSet<>(); int offset = 0; try { LogData data = getExecutionJobLog(exFlow, jobId, offset, 50000, attempt); while (data != null && data.getLength() > 0) { this.logger.info("Get application ID for execution " + exFlow.getExecutionId() + ", job" + " " + jobId + ", attempt " + attempt + ", data offset " + offset); String logData = data.getData(); final int indexOfLastSpace = logData.lastIndexOf(' '); final int indexOfLastTab = logData.lastIndexOf('\t'); final int indexOfLastEoL = logData.lastIndexOf('\n'); final int indexOfLastDelim = Math .max(indexOfLastEoL, Math.max(indexOfLastSpace, indexOfLastTab)); if (indexOfLastDelim > -1) { logData = logData.substring(0, indexOfLastDelim + 1); } applicationIds.addAll(ExecutionControllerUtils.findApplicationIdsFromLog(logData)); offset = data.getOffset() + logData.length(); data = getExecutionJobLog(exFlow, jobId, offset, 50000, attempt); } } catch (final ExecutorManagerException e) { this.logger.error("Failed to get application ID for execution " + exFlow.getExecutionId() + ", job " + jobId + ", attempt " + attempt + ", data offset " + offset, e); } return applicationIds; } @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 testGetApplicationIdsFromLog() throws Exception { when(this.loader.fetchActiveFlowByExecId(this.flow1.getExecutionId())) .thenReturn(new Pair<>(this.ref1, this.flow1)); final String logData1 = "Submitted application_12345_6789."; when(this.apiGateway.callWithReference(any(), eq(ConnectorParams.LOG_ACTION), any())) .then(getLogChunksMock(logData1)); Set<String> appIds = this.controller.getApplicationIds(this.flow1, "job1", 0); Assert.assertEquals(1, appIds.size()); Assert.assertEquals("12345_6789", appIds.iterator().next()); final String logData2 = " Submitted application_12345_6789.\n AttemptID: attempt_12345_6789. " + "Accepted application_98765_4321."; when(this.apiGateway.callWithReference(any(), eq(ConnectorParams.LOG_ACTION), any())) .then(getLogChunksMock(logData2)); appIds = this.controller.getApplicationIds(this.flow1, "job1", 0); Assert.assertEquals(2, appIds.size()); final Iterator iterator = appIds.iterator(); Assert.assertEquals("12345_6789", iterator.next()); Assert.assertEquals("98765_4321", iterator.next()); when(this.apiGateway.callWithReference(any(), eq(ConnectorParams.LOG_ACTION), any())) .then(getLogChunksMock("")); Assert.assertEquals(0, this.controller.getApplicationIds(this.flow1, "job1", 0).size()); }