JobManager implements ModeChangeListener { public synchronized void initialize() throws StateManagementException { this.validateStateTransition(ManagerState.INITIALIZED); try { log.info("Initializing job manager"); if (this.state == ManagerState.CREATED) { if (this.isSchedulerEnabled()) { if (this.schedulerFactory instanceof StdSchedulerFactory) { Properties quartzConfig = this.configuration.subset("org.quartz").toProperties(); ((StdSchedulerFactory) this.schedulerFactory).initialize(quartzConfig); } if (this.scheduler == null || this.scheduler.isShutdown()) { this.scheduler = this.schedulerFactory.getScheduler(); } this.synchronizeJobSchedule(); this.scheduler.setJobFactory(this.qrtzExecutor); this.scheduler.getListenerManager() .addTriggerListener(this.qrtzExecutor); } if (!this.receiver.isInitialized()) { this.receiver.initialize(this); } this.modeManager.registerModeChangeListener(this); if (this.modeManager.getCurrentMode() == Mode.SUSPEND) { this.suspendKeys.add(SUSPEND_KEY_TRIGGERED); } this.recoverAbandonedJobs(); } log.info("Job manager initialization complete"); this.state = ManagerState.INITIALIZED; } catch (Exception e) { String errmsg = "Unexpected exception occurred during initialization"; log.error(errmsg, e); try { if (this.scheduler != null) { this.scheduler.shutdown(); this.scheduler = null; } } catch (SchedulerException se) { log.error("Unable to shutdown quartz scheduler while processing other exceptions", se); } throw new StateManagementException(this.state, ManagerState.INITIALIZED, errmsg, e); } } @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 testJobScheduling() { String schedule = "%d * * * * ?"; int jobs = 3; for (int i = 0; i < jobs; ++i) { String cfgName = ConfigProperties.jobConfig(TestJob.JOB_KEY + '-' + i, ConfigProperties.ASYNC_JOBS_JOB_SCHEDULE); this.config.setProperty(cfgName, String.format(schedule, i)); } JobManager manager = this.createJobManager(); manager.initialize(); for (int i = 0; i < jobs; ++i) { this.verifyScheduledJob(TestJob.JOB_KEY + '-' + i, String.format(schedule, i)); } }
@Test public void testJobSchedulingDoesNotScheduleDisabledJobs() { this.config.setProperty( ConfigProperties.jobConfig(TestJob.JOB_KEY, ConfigProperties.ASYNC_JOBS_JOB_SCHEDULE), "5 * * * * ?"); this.config.setProperty( ConfigProperties.jobConfig(TestJob.JOB_KEY, ConfigProperties.ASYNC_JOBS_JOB_ENABLED), "false"); JobManager manager = this.createJobManager(); manager.initialize(); this.verifyNotScheduled(TestJob.JOB_KEY); }
@Test public void testJobSchedulingDoesNotScheduleBlacklistedJobs() { this.config.setProperty( ConfigProperties.jobConfig(TestJob.JOB_KEY, ConfigProperties.ASYNC_JOBS_JOB_SCHEDULE), "5 * * * * ?"); this.config.setProperty(ConfigProperties.ASYNC_JOBS_BLACKLIST, "a,b," + TestJob.JOB_KEY + ",d,e"); JobManager manager = this.createJobManager(); manager.initialize(); this.verifyNotScheduled(TestJob.JOB_KEY); }
@Test public void testJobSchedulingDoesNotScheduleJobsNotOnWhitelist() { this.config.setProperty( ConfigProperties.jobConfig(TestJob.JOB_KEY, ConfigProperties.ASYNC_JOBS_JOB_SCHEDULE), "5 * * * * ?"); this.config.setProperty(ConfigProperties.ASYNC_JOBS_WHITELIST, "a,b,c,d,e"); JobManager manager = this.createJobManager(); manager.initialize(); this.verifyNotScheduled(TestJob.JOB_KEY); }
@Test public void testManagerCanBeInitializedAfterCreation() { JobManager manager = this.createJobManager(); manager.initialize(); }
@Test public void testManagerCannotBeInitializedTwice() { JobManager manager = this.createJobManager(); manager.initialize(); assertThrows(IllegalStateException.class, manager::initialize); }
@Test public void testSchedulerRemovesDeadJobsOnStartup() throws Exception { StdSchedulerFactory schedulerFactory = mock(StdSchedulerFactory.class); this.schedulerFactory = schedulerFactory; doReturn(this.scheduler).when(this.schedulerFactory).getScheduler(); JobKey jobkey1 = new JobKey("job-1", "test_job_group"); JobKey jobkey2 = new JobKey("job-2", "test_job_group"); JobKey jobkey3 = new JobKey("job-3", "test_job_group"); doReturn(Util.asSet(jobkey1, jobkey2, jobkey3)).when(this.scheduler) .getJobKeys(any(GroupMatcher.class)); doThrow(new JobPersistenceException("kaboom")).when(this.scheduler).getJobDetail(eq(jobkey2)); JobManager manager = this.createJobManager(); manager.initialize(); verify(this.scheduler, times(1)).getJobDetail(eq(jobkey1)); verify(this.scheduler, times(1)).getJobDetail(eq(jobkey2)); verify(this.scheduler, times(1)).getJobDetail(eq(jobkey3)); verify(this.scheduler, times(1)).deleteJob(eq(jobkey2)); }