@Test public void testExecuteBatchRequest() throws Exception { Clusters clustersMock = createMock(Clusters.class); Cluster clusterMock = createMock(Cluster.class); RequestExecution requestExecutionMock = createMock(RequestExecution.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(); BatchRequest batchRequestMock = createMock(BatchRequest.class); long executionId = 11L; long batchId = 1L; long requestId = 5L; String clusterName = "mycluster"; String uri = "clusters"; String type = "post"; String body = "body"; Integer userId = 1; Map<Long, RequestExecution> executionMap = new HashMap<Long, RequestExecution>(); executionMap.put(executionId, requestExecutionMock); BatchRequestResponse batchRequestResponse = new BatchRequestResponse(); batchRequestResponse.setStatus(HostRoleStatus.IN_PROGRESS.toString()); batchRequestResponse.setRequestId(requestId); batchRequestResponse.setReturnCode(202); EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE); EasyMock.replay(configurationMock); ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class). withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock, actionDBAccessorMock, gson). addMockedMethods("performApiRequest", "updateBatchRequest").createNiceMock(); expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes(); expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes(); expect(requestExecutionMock.getBatchRequest(eq(batchId))).andReturn(batchRequestMock).once(); expect(requestExecutionMock.getRequestBody(eq(batchId))).andReturn(body).once(); expect(requestExecutionMock.getAuthenticatedUserId()).andReturn(userId).once(); expect(batchRequestMock.getUri()).andReturn(uri).once(); expect(batchRequestMock.getType()).andReturn(type).once(); expect(scheduleManager.performApiRequest(eq(uri), eq(body), eq(type), eq(userId))).andReturn(batchRequestResponse).once(); scheduleManager.updateBatchRequest(eq(executionId), eq(batchId), eq(clusterName), eq(batchRequestResponse), eq(false)); expectLastCall().once(); actionDBAccessorMock.setSourceScheduleForRequest(eq(requestId), eq(executionId)); expectLastCall().once(); replay(clusterMock, clustersMock, requestExecutionMock, executionSchedulerMock, tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock); scheduleManager.executeBatchRequest(executionId, batchId, clusterName); verify(clusterMock, clustersMock, configurationMock, requestExecutionMock, executionSchedulerMock, tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock); }
public Long executeBatchRequest(long executionId, long batchId, String clusterName) throws AmbariException { String type = null; String uri = null; String body = null; try { RequestExecution requestExecution = clusters.getCluster(clusterName).getAllRequestExecutions().get(executionId); BatchRequest batchRequest = requestExecution.getBatchRequest(batchId); type = batchRequest.getType(); uri = batchRequest.getUri(); body = requestExecution.getRequestBody(batchId); BatchRequestResponse batchRequestResponse = performApiRequest(uri, body, type, requestExecution.getAuthenticatedUserId()); updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, false); if (batchRequestResponse.getRequestId() != null) { actionDBAccessor.setSourceScheduleForRequest(batchRequestResponse.getRequestId(), executionId); } return batchRequestResponse.getRequestId(); } catch (Exception e) { throw new AmbariException("Exception occurred while performing request", e); } }
ExecutionScheduleManager { public Long executeBatchRequest(long executionId, long batchId, String clusterName) throws AmbariException { String type = null; String uri = null; String body = null; try { RequestExecution requestExecution = clusters.getCluster(clusterName).getAllRequestExecutions().get(executionId); BatchRequest batchRequest = requestExecution.getBatchRequest(batchId); type = batchRequest.getType(); uri = batchRequest.getUri(); body = requestExecution.getRequestBody(batchId); BatchRequestResponse batchRequestResponse = performApiRequest(uri, body, type, requestExecution.getAuthenticatedUserId()); updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, false); if (batchRequestResponse.getRequestId() != null) { actionDBAccessor.setSourceScheduleForRequest(batchRequestResponse.getRequestId(), executionId); } return batchRequestResponse.getRequestId(); } catch (Exception e) { throw new AmbariException("Exception occurred while performing request", e); } } }
ExecutionScheduleManager { public Long executeBatchRequest(long executionId, long batchId, String clusterName) throws AmbariException { String type = null; String uri = null; String body = null; try { RequestExecution requestExecution = clusters.getCluster(clusterName).getAllRequestExecutions().get(executionId); BatchRequest batchRequest = requestExecution.getBatchRequest(batchId); type = batchRequest.getType(); uri = batchRequest.getUri(); body = requestExecution.getRequestBody(batchId); BatchRequestResponse batchRequestResponse = performApiRequest(uri, body, type, requestExecution.getAuthenticatedUserId()); updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, false); if (batchRequestResponse.getRequestId() != null) { actionDBAccessor.setSourceScheduleForRequest(batchRequestResponse.getRequestId(), executionId); } return batchRequestResponse.getRequestId(); } catch (Exception e) { throw new AmbariException("Exception occurred while performing request", e); } } @Inject  ExecutionScheduleManager(Configuration configuration,
                                  ExecutionScheduler executionScheduler,
                                  InternalTokenStorage tokenStorage,
                                  Clusters clusters,
                                  ActionDBAccessor actionDBAccessor,
                                  Gson gson); }
ExecutionScheduleManager { public Long executeBatchRequest(long executionId, long batchId, String clusterName) throws AmbariException { String type = null; String uri = null; String body = null; try { RequestExecution requestExecution = clusters.getCluster(clusterName).getAllRequestExecutions().get(executionId); BatchRequest batchRequest = requestExecution.getBatchRequest(batchId); type = batchRequest.getType(); uri = batchRequest.getUri(); body = requestExecution.getRequestBody(batchId); BatchRequestResponse batchRequestResponse = performApiRequest(uri, body, type, requestExecution.getAuthenticatedUserId()); updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, false); if (batchRequestResponse.getRequestId() != null) { actionDBAccessor.setSourceScheduleForRequest(batchRequestResponse.getRequestId(), executionId); } return batchRequestResponse.getRequestId(); } catch (Exception e) { throw new AmbariException("Exception occurred while performing request", e); } } @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 Long executeBatchRequest(long executionId, long batchId, String clusterName) throws AmbariException { String type = null; String uri = null; String body = null; try { RequestExecution requestExecution = clusters.getCluster(clusterName).getAllRequestExecutions().get(executionId); BatchRequest batchRequest = requestExecution.getBatchRequest(batchId); type = batchRequest.getType(); uri = batchRequest.getUri(); body = requestExecution.getRequestBody(batchId); BatchRequestResponse batchRequestResponse = performApiRequest(uri, body, type, requestExecution.getAuthenticatedUserId()); updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, false); if (batchRequestResponse.getRequestId() != null) { actionDBAccessor.setSourceScheduleForRequest(batchRequestResponse.getRequestId(), executionId); } return batchRequestResponse.getRequestId(); } catch (Exception e) { throw new AmbariException("Exception occurred while performing request", e); } } @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; }