JobManager implements IJobManager { @Override public void cancel(JobId jobId, IResultCallback<Void> callback) throws HyracksException { if (activeRunMap.containsKey(jobId)) { JobRun jobRun = activeRunMap.get(jobId); jobRun.getExecutor().cancelJob(callback); return; } JobRun jobRun = jobQueue.remove(jobId); if (jobRun != null) { List<Exception> exceptions = Collections.singletonList(HyracksException.create(ErrorCode.JOB_CANCELED, jobId)); jobRun.setStatus(JobStatus.FAILURE_BEFORE_EXECUTION, exceptions); runMapArchive.put(jobId, jobRun); runMapHistory.put(jobId, exceptions); CCServiceContext serviceCtx = ccs.getContext(); if (serviceCtx != null) { try { serviceCtx.notifyJobFinish(jobId, JobStatus.FAILURE_BEFORE_EXECUTION, exceptions); } catch (Exception e) { LOGGER.error("Exception notifying cancel on pending job {}", jobId, e); throw HyracksDataException.create(e); } } } callback.setValue(null); } JobManager(CCConfig ccConfig, ClusterControllerService ccs, IJobCapacityController jobCapacityController); @Override void add(JobRun jobRun); @Override void cancel(JobId jobId, IResultCallback<Void> callback); @Override void prepareComplete(JobRun run, JobStatus status, List<Exception> exceptions); @Override void finalComplete(JobRun run); @Override Collection<JobRun> getRunningJobs(); @Override Collection<JobRun> getPendingJobs(); @Override Collection<JobRun> getArchivedJobs(); @Override JobRun get(JobId jobId); @Override List<Exception> getExceptionHistory(JobId jobId); @Override int getJobQueueCapacity();  }
@Test public void testCancel() throws Exception { CCConfig ccConfig = new CCConfig(); IJobCapacityController jobCapacityController = mock(IJobCapacityController.class); IJobManager jobManager = spy(new JobManager(ccConfig, mockClusterControllerService(), jobCapacityController)); List<JobRun> acceptedRuns = new ArrayList<>(); for (int id = 0; id < 4096; ++id) { JobRun run = mockJobRun(id); JobSpecification job = mock(JobSpecification.class); when(run.getJobSpecification()).thenReturn(job); when(jobCapacityController.allocate(job)).thenReturn(IJobCapacityController.JobSubmissionStatus.EXECUTE); acceptedRuns.add(run); jobManager.add(run); Assert.assertTrue(jobManager.getRunningJobs().size() == id + 1); Assert.assertTrue(jobManager.getPendingJobs().isEmpty()); } List<JobRun> deferredRuns = new ArrayList<>(); for (int id = 4096; id < 8192; ++id) { JobRun run = mockJobRun(id); JobSpecification job = mock(JobSpecification.class); when(run.getJobSpecification()).thenReturn(job); when(jobCapacityController.allocate(job)).thenReturn(IJobCapacityController.JobSubmissionStatus.QUEUE) .thenReturn(IJobCapacityController.JobSubmissionStatus.EXECUTE); deferredRuns.add(run); jobManager.add(run); Assert.assertTrue(jobManager.getRunningJobs().size() == 4096); Assert.assertTrue(jobManager.getPendingJobs().size() == id + 1 - 4096); } for (JobRun run : deferredRuns) { jobManager.cancel(run.getJobId(), NoOpCallback.INSTANCE); } for (JobRun run : acceptedRuns) { jobManager.cancel(run.getJobId(), NoOpCallback.INSTANCE); } Assert.assertTrue(jobManager.getPendingJobs().isEmpty()); Assert.assertTrue(jobManager.getArchivedJobs().size() == ccConfig.getJobHistorySize()); verify(jobManager, times(0)).prepareComplete(any(), any(), any()); verify(jobManager, times(0)).finalComplete(any()); }