JobManager implements ModeChangeListener { @Transactional public int abortNonTerminalJobs(AsyncJobStatusQueryBuilder queryBuilder) { if (queryBuilder == null) { queryBuilder = new AsyncJobStatusQueryBuilder(); } Collection<JobState> states = queryBuilder.getJobStates(); Stream<JobState> stateStream = states != null && !states.isEmpty() ? states.stream() : Arrays.stream(JobState.values()); states = stateStream.filter(state -> state != null && !state.isTerminal()) .collect(Collectors.toSet()); for (JobState state : states) { if (!state.isValidTransition(JobState.ABORTED)) { throw new IllegalStateException(String.format("Cannot transition from state %s to %s", state.name(), JobState.ABORTED.name())); } } queryBuilder.setJobStates(states); return this.jobCurator.updateJobState(queryBuilder, JobState.ABORTED); } @Inject  JobManager(
        Configuration configuration,
        SchedulerFactory schedulerFactory,
        CandlepinModeManager modeManager,
        AsyncJobStatusCurator jobCurator,
        OwnerCurator ownerCurator,
        JobMessageDispatcher dispatcher,
        JobMessageReceiver receiver,
        PrincipalProvider principalProvider,
        CandlepinRequestScope scope,
        Injector injector); static Class<? extends AsyncJob> registerJob(String jobKey, Class<? extends AsyncJob> jobClass); static Class<? extends AsyncJob> unregisterJob(String jobKey); static Class<? extends AsyncJob> getJobClass(String jobKey); synchronized void initialize(); void start(); synchronized void start(Object key); void resume(); void resume(Object key); void suspend(); synchronized void suspend(Object key); synchronized void shutdown(); synchronized ManagerState getManagerState(); boolean isJobEnabled(String jobKey); boolean isSchedulerEnabled(); AsyncJobStatus findJob(String jobId); List<AsyncJobStatus> findJobs(AsyncJobStatusQueryBuilder queryBuilder); @Transactional AsyncJobStatus queueJob(JobConfig config); AsyncJobStatus executeJob(JobMessage message); @Transactional AsyncJobStatus cancelJob(String jobId); @Transactional int cleanupJobs(AsyncJobStatusQueryBuilder queryBuilder); @Transactional int abortNonTerminalJobs(AsyncJobStatusQueryBuilder queryBuilder); @Override void handleModeChange(CandlepinModeManager modeManager, Mode previousMode, Mode currentMode);  }
@Test public void testAbortNonTerminalJobs() { Set<JobState> states = Arrays.stream(JobState.values()) .filter(state -> !state.isTerminal()) .collect(Collectors.toSet()); AsyncJobStatusQueryBuilder input = this.generateQueryBuilder() .setJobStates(states); int expected = new Random().nextInt(); ArgumentCaptor<AsyncJobStatusQueryBuilder> captor = ArgumentCaptor.forClass(AsyncJobStatusQueryBuilder.class); doReturn(expected).when(this.jobCurator).updateJobState(eq(input), eq(JobState.ABORTED)); JobManager manager = this.createJobManager(); int output = manager.abortNonTerminalJobs(input); assertEquals(expected, output); verify(this.jobCurator, times(1)).updateJobState(captor.capture(), eq(JobState.ABORTED)); AsyncJobStatusQueryBuilder captured = captor.getValue(); assertNotNull(captured); assertEquals(input, captured); }
@Test public void testAbortNonTerminalJobsLimitsJobStatesToNonTerminalStates() { Set<JobState> jobStates = Util.asSet(JobState.values()); AsyncJobStatusQueryBuilder input = this.generateQueryBuilder() .setJobStates(jobStates); int expected = new Random().nextInt(); ArgumentCaptor<AsyncJobStatusQueryBuilder> captor = ArgumentCaptor.forClass(AsyncJobStatusQueryBuilder.class); doReturn(expected).when(this.jobCurator).updateJobState(eq(input), eq(JobState.ABORTED)); JobManager manager = this.createJobManager(); int output = manager.abortNonTerminalJobs(input); assertEquals(expected, output); verify(this.jobCurator, times(1)).updateJobState(captor.capture(), eq(JobState.ABORTED)); AsyncJobStatusQueryBuilder captured = captor.getValue(); assertNotNull(captured); Collection<JobState> actual = captured.getJobStates(); assertNotNull(actual); assertThat(actual, not(empty())); assertThat(actual.size(), lessThan(jobStates.size())); for (JobState state : actual) { assertNotNull(state); assertFalse(state.isTerminal()); } }
@Test public void testAbortNonTerminalJobsDefaultsToNoneTerminalStates() { ArgumentCaptor<AsyncJobStatusQueryBuilder> captor = ArgumentCaptor.forClass(AsyncJobStatusQueryBuilder.class); JobManager manager = this.createJobManager(); int output = manager.abortNonTerminalJobs(null); verify(this.jobCurator, times(1)).updateJobState(captor.capture(), eq(JobState.ABORTED)); AsyncJobStatusQueryBuilder captured = captor.getValue(); assertNotNull(captured); Collection<JobState> actual = captured.getJobStates(); Set<JobState> expected = Arrays.stream(JobState.values()) .filter(state -> !state.isTerminal()) .collect(Collectors.toSet()); assertThat(expected, not(empty())); assertNotNull(actual); assertEquals(expected.size(), actual.size()); for (JobState state : expected) { assertThat(actual, hasItem(state)); } }