@SuppressWarnings("unchecked") @Test public void testFinalizeBatch() throws Exception { Clusters clustersMock = createMock(Clusters.class); Cluster clusterMock = createMock(Cluster.class); Configuration configurationMock = createNiceMock(Configuration.class); ExecutionScheduler executionSchedulerMock = createMock(ExecutionScheduler.class); InternalTokenStorage tokenStorageMock = createMock(InternalTokenStorage.class); ActionDBAccessor actionDBAccessorMock = createMock(ActionDBAccessor.class); Gson gson = new Gson(); RequestExecution requestExecutionMock = createMock(RequestExecution.class); Batch batchMock = createMock(Batch.class); JobDetail jobDetailMock = createMock(JobDetail.class); final BatchRequest batchRequestMock = createMock(BatchRequest.class); final Trigger triggerMock = createNiceMock(Trigger.class); final List<Trigger> triggers = new ArrayList<Trigger>() {{ add(triggerMock); }}; long executionId = 11L; String clusterName = "c1"; Date pastDate = new Date(new Date().getTime() - 2); Map<Long, RequestExecution> executionMap = new HashMap<Long, RequestExecution>(); executionMap.put(executionId, requestExecutionMock); EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE); EasyMock.replay(configurationMock); ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class) .withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock, actionDBAccessorMock, gson).createMock(); expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes(); expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes(); expect(requestExecutionMock.getBatch()).andReturn(batchMock).anyTimes(); expect(batchMock.getBatchRequests()).andReturn (new ArrayList<BatchRequest>() {{ add(batchRequestMock); }}); expect(batchRequestMock.getOrderId()).andReturn(1L).anyTimes(); expect(executionSchedulerMock.getJobDetail((JobKey) anyObject())) .andReturn(jobDetailMock).anyTimes(); expect((List<Trigger>) executionSchedulerMock .getTriggersForJob((JobKey) anyObject())).andReturn(triggers).anyTimes(); expect(triggerMock.mayFireAgain()).andReturn(true).anyTimes(); expect(triggerMock.getFinalFireTime()).andReturn(pastDate).anyTimes(); requestExecutionMock.updateStatus(RequestExecution.Status.COMPLETED); expectLastCall(); replay(clustersMock, clusterMock, requestExecutionMock, executionSchedulerMock, scheduleManager, batchMock, batchRequestMock, triggerMock, jobDetailMock, actionDBAccessorMock); scheduleManager.finalizeBatch(executionId, clusterName); verify(clustersMock, clusterMock, configurationMock, requestExecutionMock, executionSchedulerMock, scheduleManager, batchMock, batchRequestMock, triggerMock, jobDetailMock, actionDBAccessorMock); }
public void finalizeBatch(long executionId, String clusterName) throws AmbariException { Cluster cluster = clusters.getCluster(clusterName); RequestExecution requestExecution = cluster.getAllRequestExecutions().get(executionId); if (requestExecution == null) { throw new AmbariException("Unable to find request schedule with id = " + executionId); } Batch batch = requestExecution.getBatch(); BatchRequest firstBatchRequest = null; if (batch != null) { List<BatchRequest> batchRequests = batch.getBatchRequests(); if (batchRequests != null && batchRequests.size() > 0) { Collections.sort(batchRequests); firstBatchRequest = batchRequests.get(0); } } boolean markCompleted = false; if (firstBatchRequest != null) { String jobName = getJobName(executionId, firstBatchRequest.getOrderId()); JobKey jobKey = JobKey.jobKey(jobName, ExecutionJob.LINEAR_EXECUTION_JOB_GROUP); JobDetail jobDetail; try { jobDetail = executionScheduler.getJobDetail(jobKey); } catch (SchedulerException e) { LOG.warn("Unable to retrieve job details from scheduler. job: " + jobKey); e.printStackTrace(); return; } if (jobDetail != null) { try { List<? extends Trigger> triggers = executionScheduler.getTriggersForJob(jobKey); if (triggers != null && triggers.size() > 0) { if (triggers.size() > 1) { throw new AmbariException("Too many triggers defined for job. " + "job: " + jobKey); } Trigger trigger = triggers.get(0); if (!trigger.mayFireAgain() || (trigger.getFinalFireTime() != null && !DateUtils.isFutureTime(trigger.getFinalFireTime()))) { markCompleted = true; } } else { markCompleted = true; } } catch (SchedulerException e) { LOG.warn("Unable to retrieve triggers for job: " + jobKey); e.printStackTrace(); return; } } } if (markCompleted) { requestExecution.updateStatus(RequestExecution.Status.COMPLETED); } }
ExecutionScheduleManager { public void finalizeBatch(long executionId, String clusterName) throws AmbariException { Cluster cluster = clusters.getCluster(clusterName); RequestExecution requestExecution = cluster.getAllRequestExecutions().get(executionId); if (requestExecution == null) { throw new AmbariException("Unable to find request schedule with id = " + executionId); } Batch batch = requestExecution.getBatch(); BatchRequest firstBatchRequest = null; if (batch != null) { List<BatchRequest> batchRequests = batch.getBatchRequests(); if (batchRequests != null && batchRequests.size() > 0) { Collections.sort(batchRequests); firstBatchRequest = batchRequests.get(0); } } boolean markCompleted = false; if (firstBatchRequest != null) { String jobName = getJobName(executionId, firstBatchRequest.getOrderId()); JobKey jobKey = JobKey.jobKey(jobName, ExecutionJob.LINEAR_EXECUTION_JOB_GROUP); JobDetail jobDetail; try { jobDetail = executionScheduler.getJobDetail(jobKey); } catch (SchedulerException e) { LOG.warn("Unable to retrieve job details from scheduler. job: " + jobKey); e.printStackTrace(); return; } if (jobDetail != null) { try { List<? extends Trigger> triggers = executionScheduler.getTriggersForJob(jobKey); if (triggers != null && triggers.size() > 0) { if (triggers.size() > 1) { throw new AmbariException("Too many triggers defined for job. " + "job: " + jobKey); } Trigger trigger = triggers.get(0); if (!trigger.mayFireAgain() || (trigger.getFinalFireTime() != null && !DateUtils.isFutureTime(trigger.getFinalFireTime()))) { markCompleted = true; } } else { markCompleted = true; } } catch (SchedulerException e) { LOG.warn("Unable to retrieve triggers for job: " + jobKey); e.printStackTrace(); return; } } } if (markCompleted) { requestExecution.updateStatus(RequestExecution.Status.COMPLETED); } } }
ExecutionScheduleManager { public void finalizeBatch(long executionId, String clusterName) throws AmbariException { Cluster cluster = clusters.getCluster(clusterName); RequestExecution requestExecution = cluster.getAllRequestExecutions().get(executionId); if (requestExecution == null) { throw new AmbariException("Unable to find request schedule with id = " + executionId); } Batch batch = requestExecution.getBatch(); BatchRequest firstBatchRequest = null; if (batch != null) { List<BatchRequest> batchRequests = batch.getBatchRequests(); if (batchRequests != null && batchRequests.size() > 0) { Collections.sort(batchRequests); firstBatchRequest = batchRequests.get(0); } } boolean markCompleted = false; if (firstBatchRequest != null) { String jobName = getJobName(executionId, firstBatchRequest.getOrderId()); JobKey jobKey = JobKey.jobKey(jobName, ExecutionJob.LINEAR_EXECUTION_JOB_GROUP); JobDetail jobDetail; try { jobDetail = executionScheduler.getJobDetail(jobKey); } catch (SchedulerException e) { LOG.warn("Unable to retrieve job details from scheduler. job: " + jobKey); e.printStackTrace(); return; } if (jobDetail != null) { try { List<? extends Trigger> triggers = executionScheduler.getTriggersForJob(jobKey); if (triggers != null && triggers.size() > 0) { if (triggers.size() > 1) { throw new AmbariException("Too many triggers defined for job. " + "job: " + jobKey); } Trigger trigger = triggers.get(0); if (!trigger.mayFireAgain() || (trigger.getFinalFireTime() != null && !DateUtils.isFutureTime(trigger.getFinalFireTime()))) { markCompleted = true; } } else { markCompleted = true; } } catch (SchedulerException e) { LOG.warn("Unable to retrieve triggers for job: " + jobKey); e.printStackTrace(); return; } } } if (markCompleted) { requestExecution.updateStatus(RequestExecution.Status.COMPLETED); } } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); }
ExecutionScheduleManager { public void finalizeBatch(long executionId, String clusterName) throws AmbariException { Cluster cluster = clusters.getCluster(clusterName); RequestExecution requestExecution = cluster.getAllRequestExecutions().get(executionId); if (requestExecution == null) { throw new AmbariException("Unable to find request schedule with id = " + executionId); } Batch batch = requestExecution.getBatch(); BatchRequest firstBatchRequest = null; if (batch != null) { List<BatchRequest> batchRequests = batch.getBatchRequests(); if (batchRequests != null && batchRequests.size() > 0) { Collections.sort(batchRequests); firstBatchRequest = batchRequests.get(0); } } boolean markCompleted = false; if (firstBatchRequest != null) { String jobName = getJobName(executionId, firstBatchRequest.getOrderId()); JobKey jobKey = JobKey.jobKey(jobName, ExecutionJob.LINEAR_EXECUTION_JOB_GROUP); JobDetail jobDetail; try { jobDetail = executionScheduler.getJobDetail(jobKey); } catch (SchedulerException e) { LOG.warn("Unable to retrieve job details from scheduler. job: " + jobKey); e.printStackTrace(); return; } if (jobDetail != null) { try { List<? extends Trigger> triggers = executionScheduler.getTriggersForJob(jobKey); if (triggers != null && triggers.size() > 0) { if (triggers.size() > 1) { throw new AmbariException("Too many triggers defined for job. " + "job: " + jobKey); } Trigger trigger = triggers.get(0); if (!trigger.mayFireAgain() || (trigger.getFinalFireTime() != null && !DateUtils.isFutureTime(trigger.getFinalFireTime()))) { markCompleted = true; } } else { markCompleted = true; } } catch (SchedulerException e) { LOG.warn("Unable to retrieve triggers for job: " + jobKey); e.printStackTrace(); return; } } } if (markCompleted) { requestExecution.updateStatus(RequestExecution.Status.COMPLETED); } } @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 finalizeBatch(long executionId, String clusterName) throws AmbariException { Cluster cluster = clusters.getCluster(clusterName); RequestExecution requestExecution = cluster.getAllRequestExecutions().get(executionId); if (requestExecution == null) { throw new AmbariException("Unable to find request schedule with id = " + executionId); } Batch batch = requestExecution.getBatch(); BatchRequest firstBatchRequest = null; if (batch != null) { List<BatchRequest> batchRequests = batch.getBatchRequests(); if (batchRequests != null && batchRequests.size() > 0) { Collections.sort(batchRequests); firstBatchRequest = batchRequests.get(0); } } boolean markCompleted = false; if (firstBatchRequest != null) { String jobName = getJobName(executionId, firstBatchRequest.getOrderId()); JobKey jobKey = JobKey.jobKey(jobName, ExecutionJob.LINEAR_EXECUTION_JOB_GROUP); JobDetail jobDetail; try { jobDetail = executionScheduler.getJobDetail(jobKey); } catch (SchedulerException e) { LOG.warn("Unable to retrieve job details from scheduler. job: " + jobKey); e.printStackTrace(); return; } if (jobDetail != null) { try { List<? extends Trigger> triggers = executionScheduler.getTriggersForJob(jobKey); if (triggers != null && triggers.size() > 0) { if (triggers.size() > 1) { throw new AmbariException("Too many triggers defined for job. " + "job: " + jobKey); } Trigger trigger = triggers.get(0); if (!trigger.mayFireAgain() || (trigger.getFinalFireTime() != null && !DateUtils.isFutureTime(trigger.getFinalFireTime()))) { markCompleted = true; } } else { markCompleted = true; } } catch (SchedulerException e) { LOG.warn("Unable to retrieve triggers for job: " + jobKey); e.printStackTrace(); return; } } } if (markCompleted) { requestExecution.updateStatus(RequestExecution.Status.COMPLETED); } } @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; }