@Test public void testHasToleranceThresholdExceeded() 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); long executionId = 11L; String clusterName = "c1"; BatchSettings batchSettings = new BatchSettings(); batchSettings.setTaskFailureToleranceLimit(1); Map<Long, RequestExecution> executionMap = new HashMap<Long, RequestExecution>(); executionMap.put(executionId, requestExecutionMock); expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes(); expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes(); expect(requestExecutionMock.getBatch()).andReturn(batchMock).anyTimes(); expect(batchMock.getBatchSettings()).andReturn(batchSettings).anyTimes(); replay(clustersMock, clusterMock, configurationMock, requestExecutionMock, executionSchedulerMock, batchMock); ExecutionScheduleManager scheduleManager = new ExecutionScheduleManager(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock, actionDBAccessorMock, gson); HashMap<String, Integer> taskCounts = new HashMap<String, Integer>() {{ put(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY, 2); put(BatchRequestJob.BATCH_REQUEST_TOTAL_TASKS_KEY, 10); }}; boolean exceeded = scheduleManager.hasToleranceThresholdExceeded (executionId, clusterName, taskCounts); Assert.assertTrue(exceeded); verify(clustersMock, clusterMock, configurationMock, requestExecutionMock, executionSchedulerMock, batchMock); }
public boolean hasToleranceThresholdExceeded(Long executionId, String clusterName, Map<String, Integer> taskCounts) 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); } BatchSettings batchSettings = requestExecution.getBatch().getBatchSettings(); if (batchSettings != null && batchSettings.getTaskFailureToleranceLimit() != null) { return taskCounts.get(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY) > batchSettings.getTaskFailureToleranceLimit(); } return false; }
ExecutionScheduleManager { public boolean hasToleranceThresholdExceeded(Long executionId, String clusterName, Map<String, Integer> taskCounts) 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); } BatchSettings batchSettings = requestExecution.getBatch().getBatchSettings(); if (batchSettings != null && batchSettings.getTaskFailureToleranceLimit() != null) { return taskCounts.get(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY) > batchSettings.getTaskFailureToleranceLimit(); } return false; } }
ExecutionScheduleManager { public boolean hasToleranceThresholdExceeded(Long executionId, String clusterName, Map<String, Integer> taskCounts) 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); } BatchSettings batchSettings = requestExecution.getBatch().getBatchSettings(); if (batchSettings != null && batchSettings.getTaskFailureToleranceLimit() != null) { return taskCounts.get(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY) > batchSettings.getTaskFailureToleranceLimit(); } return false; } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); }
ExecutionScheduleManager { public boolean hasToleranceThresholdExceeded(Long executionId, String clusterName, Map<String, Integer> taskCounts) 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); } BatchSettings batchSettings = requestExecution.getBatch().getBatchSettings(); if (batchSettings != null && batchSettings.getTaskFailureToleranceLimit() != null) { return taskCounts.get(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY) > batchSettings.getTaskFailureToleranceLimit(); } return false; } @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 boolean hasToleranceThresholdExceeded(Long executionId, String clusterName, Map<String, Integer> taskCounts) 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); } BatchSettings batchSettings = requestExecution.getBatch().getBatchSettings(); if (batchSettings != null && batchSettings.getTaskFailureToleranceLimit() != null) { return taskCounts.get(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY) > batchSettings.getTaskFailureToleranceLimit(); } return false; } @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; }