@Test public void testScheduleBatch() throws Exception { RequestExecution requestExecution = createRequestExecution(true); Assert.assertNotNull(requestExecution); executionScheduleManager.scheduleBatch(requestExecution); String jobName1 = executionScheduleManager.getJobName(requestExecution .getId(), 10L); String jobName2 = executionScheduleManager.getJobName(requestExecution .getId(), 12L); JobDetail jobDetail1 = null; JobDetail jobDetail2 = null; Trigger trigger1 = null; Trigger trigger2 = null; for(String group: scheduler.getJobGroupNames()) { for(JobKey jobKey : scheduler.getJobKeys(GroupMatcher.jobGroupEquals (ExecutionJob.LINEAR_EXECUTION_JOB_GROUP))) { LOG.info("Found job identified by: " + jobKey); String jobName = jobKey.getName(); String jobGroup = jobKey.getGroup(); List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(jobKey); Trigger trigger = triggers != null && !triggers.isEmpty() ? triggers.get(0) : null; Date nextFireTime = trigger != null ? trigger.getNextFireTime() : null; LOG.info("[jobName] : " + jobName + " [groupName] : " + jobGroup + " - " + nextFireTime); if (jobName.equals(jobName1)) { jobDetail1 = scheduler.getJobDetail(jobKey); trigger1 = trigger; } else if (jobName.equals(jobName2)) { jobDetail2 = scheduler.getJobDetail(jobKey); trigger2 = trigger; } } } Assert.assertNotNull(jobDetail1); Assert.assertNotNull(trigger1); Assert.assertNotNull(jobDetail2); Assert.assertNull(trigger2); CronTrigger cronTrigger = (CronTrigger) trigger1; Schedule schedule = new Schedule(); schedule.setMinutes("10"); schedule.setHours("2"); schedule.setMonth("*"); schedule.setDaysOfMonth("*"); schedule.setDayOfWeek("?"); Assert.assertEquals(schedule.getScheduleExpression(), cronTrigger.getCronExpression()); Assert.assertEquals(jobName1, jobDetail1.getKey().getName()); Assert.assertEquals(jobName2, jobDetail2.getKey().getName()); }
public void scheduleBatch(RequestExecution requestExecution) throws AmbariException { if (!isSchedulerAvailable()) { throw new AmbariException("Scheduler unavailable."); } try { if (!executionScheduler.isSchedulerStarted()) { executionScheduler.startScheduler(null); } } catch (SchedulerException e) { LOG.error("Unable to determine scheduler state.", e); throw new AmbariException("Scheduler unavailable."); } JobDetail firstJobDetail = persistBatch(requestExecution); if (firstJobDetail == null) { throw new AmbariException("Unable to schedule jobs. firstJobDetail = " + firstJobDetail); } Schedule schedule = requestExecution.getSchedule(); if (schedule != null) { String triggerExpression = schedule.getScheduleExpression(); Date startDate = null; Date endDate = null; try { String startTime = schedule.getStartTime(); String endTime = schedule.getEndTime(); startDate = startTime != null && !startTime.isEmpty() ? DateUtils.convertToDate(startTime) : new Date(); endDate = endTime != null && !endTime.isEmpty() ? DateUtils.convertToDate(endTime) : null; } catch (ParseException e) { LOG.error("Unable to parse startTime / endTime.", e); } Trigger trigger = newTrigger() .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(cronSchedule(triggerExpression) .withMisfireHandlingInstructionFireAndProceed()) .forJob(firstJobDetail) .startAt(startDate) .endAt(endDate) .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } else { Trigger trigger = newTrigger() .forJob(firstJobDetail) .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(simpleSchedule().withMisfireHandlingInstructionFireNow()) .startNow() .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } }
ExecutionScheduleManager { public void scheduleBatch(RequestExecution requestExecution) throws AmbariException { if (!isSchedulerAvailable()) { throw new AmbariException("Scheduler unavailable."); } try { if (!executionScheduler.isSchedulerStarted()) { executionScheduler.startScheduler(null); } } catch (SchedulerException e) { LOG.error("Unable to determine scheduler state.", e); throw new AmbariException("Scheduler unavailable."); } JobDetail firstJobDetail = persistBatch(requestExecution); if (firstJobDetail == null) { throw new AmbariException("Unable to schedule jobs. firstJobDetail = " + firstJobDetail); } Schedule schedule = requestExecution.getSchedule(); if (schedule != null) { String triggerExpression = schedule.getScheduleExpression(); Date startDate = null; Date endDate = null; try { String startTime = schedule.getStartTime(); String endTime = schedule.getEndTime(); startDate = startTime != null && !startTime.isEmpty() ? DateUtils.convertToDate(startTime) : new Date(); endDate = endTime != null && !endTime.isEmpty() ? DateUtils.convertToDate(endTime) : null; } catch (ParseException e) { LOG.error("Unable to parse startTime / endTime.", e); } Trigger trigger = newTrigger() .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(cronSchedule(triggerExpression) .withMisfireHandlingInstructionFireAndProceed()) .forJob(firstJobDetail) .startAt(startDate) .endAt(endDate) .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } else { Trigger trigger = newTrigger() .forJob(firstJobDetail) .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(simpleSchedule().withMisfireHandlingInstructionFireNow()) .startNow() .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } } }
ExecutionScheduleManager { public void scheduleBatch(RequestExecution requestExecution) throws AmbariException { if (!isSchedulerAvailable()) { throw new AmbariException("Scheduler unavailable."); } try { if (!executionScheduler.isSchedulerStarted()) { executionScheduler.startScheduler(null); } } catch (SchedulerException e) { LOG.error("Unable to determine scheduler state.", e); throw new AmbariException("Scheduler unavailable."); } JobDetail firstJobDetail = persistBatch(requestExecution); if (firstJobDetail == null) { throw new AmbariException("Unable to schedule jobs. firstJobDetail = " + firstJobDetail); } Schedule schedule = requestExecution.getSchedule(); if (schedule != null) { String triggerExpression = schedule.getScheduleExpression(); Date startDate = null; Date endDate = null; try { String startTime = schedule.getStartTime(); String endTime = schedule.getEndTime(); startDate = startTime != null && !startTime.isEmpty() ? DateUtils.convertToDate(startTime) : new Date(); endDate = endTime != null && !endTime.isEmpty() ? DateUtils.convertToDate(endTime) : null; } catch (ParseException e) { LOG.error("Unable to parse startTime / endTime.", e); } Trigger trigger = newTrigger() .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(cronSchedule(triggerExpression) .withMisfireHandlingInstructionFireAndProceed()) .forJob(firstJobDetail) .startAt(startDate) .endAt(endDate) .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } else { Trigger trigger = newTrigger() .forJob(firstJobDetail) .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(simpleSchedule().withMisfireHandlingInstructionFireNow()) .startNow() .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); }
ExecutionScheduleManager { public void scheduleBatch(RequestExecution requestExecution) throws AmbariException { if (!isSchedulerAvailable()) { throw new AmbariException("Scheduler unavailable."); } try { if (!executionScheduler.isSchedulerStarted()) { executionScheduler.startScheduler(null); } } catch (SchedulerException e) { LOG.error("Unable to determine scheduler state.", e); throw new AmbariException("Scheduler unavailable."); } JobDetail firstJobDetail = persistBatch(requestExecution); if (firstJobDetail == null) { throw new AmbariException("Unable to schedule jobs. firstJobDetail = " + firstJobDetail); } Schedule schedule = requestExecution.getSchedule(); if (schedule != null) { String triggerExpression = schedule.getScheduleExpression(); Date startDate = null; Date endDate = null; try { String startTime = schedule.getStartTime(); String endTime = schedule.getEndTime(); startDate = startTime != null && !startTime.isEmpty() ? DateUtils.convertToDate(startTime) : new Date(); endDate = endTime != null && !endTime.isEmpty() ? DateUtils.convertToDate(endTime) : null; } catch (ParseException e) { LOG.error("Unable to parse startTime / endTime.", e); } Trigger trigger = newTrigger() .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(cronSchedule(triggerExpression) .withMisfireHandlingInstructionFireAndProceed()) .forJob(firstJobDetail) .startAt(startDate) .endAt(endDate) .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } else { Trigger trigger = newTrigger() .forJob(firstJobDetail) .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(simpleSchedule().withMisfireHandlingInstructionFireNow()) .startNow() .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); void start(); void stop(); boolean isSchedulerAvailable(); void scheduleJob(Trigger trigger); boolean continueOnMisfire(JobExecutionContext jobExecutionContext); void scheduleBatch(RequestExecution requestExecution); void updateBatchSchedule(RequestExecution requestExecution); void validateSchedule(Schedule schedule); void deleteAllJobs(RequestExecution requestExecution); Long executeBatchRequest(long executionId,
                                  long batchId,
                                  String clusterName); BatchRequestResponse getBatchRequestResponse(Long requestId, String clusterName); void updateBatchRequest(long executionId, long batchId, String clusterName,
                                 BatchRequestResponse batchRequestResponse,
                                 boolean statusOnly); boolean hasToleranceThresholdExceeded(Long executionId,
      String clusterName, Map<String, Integer> taskCounts); void finalizeBatch(long executionId, String clusterName); }
ExecutionScheduleManager { public void scheduleBatch(RequestExecution requestExecution) throws AmbariException { if (!isSchedulerAvailable()) { throw new AmbariException("Scheduler unavailable."); } try { if (!executionScheduler.isSchedulerStarted()) { executionScheduler.startScheduler(null); } } catch (SchedulerException e) { LOG.error("Unable to determine scheduler state.", e); throw new AmbariException("Scheduler unavailable."); } JobDetail firstJobDetail = persistBatch(requestExecution); if (firstJobDetail == null) { throw new AmbariException("Unable to schedule jobs. firstJobDetail = " + firstJobDetail); } Schedule schedule = requestExecution.getSchedule(); if (schedule != null) { String triggerExpression = schedule.getScheduleExpression(); Date startDate = null; Date endDate = null; try { String startTime = schedule.getStartTime(); String endTime = schedule.getEndTime(); startDate = startTime != null && !startTime.isEmpty() ? DateUtils.convertToDate(startTime) : new Date(); endDate = endTime != null && !endTime.isEmpty() ? DateUtils.convertToDate(endTime) : null; } catch (ParseException e) { LOG.error("Unable to parse startTime / endTime.", e); } Trigger trigger = newTrigger() .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(cronSchedule(triggerExpression) .withMisfireHandlingInstructionFireAndProceed()) .forJob(firstJobDetail) .startAt(startDate) .endAt(endDate) .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } else { Trigger trigger = newTrigger() .forJob(firstJobDetail) .withIdentity(REQUEST_EXECUTION_TRIGGER_PREFIX + "-" + requestExecution.getId(), ExecutionJob.LINEAR_EXECUTION_TRIGGER_GROUP) .withSchedule(simpleSchedule().withMisfireHandlingInstructionFireNow()) .startNow() .build(); try { executionScheduler.scheduleJob(trigger); LOG.debug("Scheduled trigger next fire time: " + trigger.getNextFireTime()); } catch (SchedulerException e) { LOG.error("Unable to schedule request execution.", e); throw new AmbariException(e.getMessage()); } } } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); void start(); void stop(); boolean isSchedulerAvailable(); void scheduleJob(Trigger trigger); boolean continueOnMisfire(JobExecutionContext jobExecutionContext); void scheduleBatch(RequestExecution requestExecution); void updateBatchSchedule(RequestExecution requestExecution); void validateSchedule(Schedule schedule); void deleteAllJobs(RequestExecution requestExecution); Long executeBatchRequest(long executionId,
                                  long batchId,
                                  String clusterName); BatchRequestResponse getBatchRequestResponse(Long requestId, String clusterName); void updateBatchRequest(long executionId, long batchId, String clusterName,
                                 BatchRequestResponse batchRequestResponse,
                                 boolean statusOnly); boolean hasToleranceThresholdExceeded(Long executionId,
      String clusterName, Map<String, Integer> taskCounts); void finalizeBatch(long executionId, String clusterName); static final String USER_ID_HEADER; }