You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-issues@hadoop.apache.org by GitBox <gi...@apache.org> on 2022/12/07 16:12:11 UTC
[GitHub] [hadoop] goiri commented on a diff in pull request #5175: YARN-11226. [Federation] Add createNewReservation, submitReservation, updateReservation, deleteReservation REST APIs for Router.
goiri commented on code in PR #5175:
URL: https://github.com/apache/hadoop/pull/5175#discussion_r1042404846
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/webapp/MockDefaultRequestInterceptorREST.java:
##########
@@ -859,44 +878,191 @@ public Response listReservation(String queue, String reservationId, long startTi
" Please try again with a valid reservable queue.");
}
- MockRM mockRM = setupResourceManager();
+ ReservationId reservationID =
+ ReservationId.parseReservationId(reservationId);
- ReservationId reservationID = ReservationId.parseReservationId(reservationId);
- ReservationSystem reservationSystem = mockRM.getReservationSystem();
- reservationSystem.synchronizePlan(QUEUE_DEDICATED_FULL, true);
+ if (!reservationMap.containsKey(reservationID)) {
+ throw new NotFoundException("reservationId with id: " + reservationId + " not found");
+ }
- // Generate reserved resources
ClientRMService clientService = mockRM.getClientRMService();
- // arrival time from which the resource(s) can be allocated.
- long arrival = Time.now();
-
- // deadline by when the resource(s) must be allocated.
- // The reason for choosing 1.05 is because this gives an integer
- // DURATION * 0.05 = 3000(ms)
- // deadline = arrival + 3000ms
- long deadline = (long) (arrival + 1.05 * DURATION);
-
- // In this test of reserved resources, we will apply for 4 containers (1 core, 1GB memory)
- // arrival = Time.now(), and make sure deadline - arrival > duration,
- // the current setting is greater than 3000ms
- ReservationSubmissionRequest submissionRequest =
- ReservationSystemTestUtil.createSimpleReservationRequest(
- reservationID, NUM_CONTAINERS, arrival, deadline, DURATION);
- clientService.submitReservation(submissionRequest);
-
// listReservations
ReservationListRequest request = ReservationListRequest.newInstance(
- queue, reservationID.toString(), startTime, endTime, includeResourceAllocations);
+ queue, reservationId, startTime, endTime, includeResourceAllocations);
ReservationListResponse resRespInfo = clientService.listReservations(request);
ReservationListInfo resResponse =
new ReservationListInfo(resRespInfo, includeResourceAllocations);
- if (mockRM != null) {
- mockRM.stop();
+ return Response.status(Status.OK).entity(resResponse).build();
+ }
+
+ @Override
+ public Response createNewReservation(HttpServletRequest hsr)
+ throws AuthorizationException, IOException, InterruptedException {
+
+ if (!isRunning) {
+ throw new RuntimeException("RM is stopped");
+ }
+
+ ReservationId resId = ReservationId.newInstance(Time.now(), resCounter.incrementAndGet());
+ LOG.info("Allocated new reservationId: {}.", resId);
+
+ NewReservation reservationId = new NewReservation(resId.toString());
+ return Response.status(Status.OK).entity(reservationId).build();
+ }
+
+ @Override
+ public Response submitReservation(ReservationSubmissionRequestInfo resContext,
+ HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException {
+
+ if (!isRunning) {
+ throw new RuntimeException("RM is stopped");
}
- return Response.status(Status.OK).entity(resResponse).build();
+ ReservationId reservationId = ReservationId.parseReservationId(resContext.getReservationId());
+ ReservationDefinitionInfo definitionInfo = resContext.getReservationDefinition();
+ ReservationDefinition definition =
+ RouterServerUtil.convertReservationDefinition(definitionInfo);
+ ReservationSubmissionRequest request = ReservationSubmissionRequest.newInstance(
+ definition, resContext.getQueue(), reservationId);
+ submitReservation(request);
+
+ LOG.info("Reservation submitted: {}.", reservationId);
+
+ SubClusterId subClusterId = getSubClusterId();
+ reservationMap.put(reservationId, subClusterId);
+
+ return Response.status(Status.ACCEPTED).build();
+ }
+
+ private void submitReservation(ReservationSubmissionRequest request) {
+ try {
+ // synchronize plan
+ ReservationSystem reservationSystem = mockRM.getReservationSystem();
+ reservationSystem.synchronizePlan(QUEUE_DEDICATED_FULL, true);
+ // Generate reserved resources
+ ClientRMService clientService = mockRM.getClientRMService();
+ clientService.submitReservation(request);
+ } catch (IOException | YarnException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Response updateReservation(ReservationUpdateRequestInfo resContext,
+ HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException {
+
+ if (resContext == null || resContext.getReservationId() == null ||
+ resContext.getReservationDefinition() == null) {
Review Comment:
indentation
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/webapp/MockDefaultRequestInterceptorREST.java:
##########
@@ -859,44 +878,191 @@ public Response listReservation(String queue, String reservationId, long startTi
" Please try again with a valid reservable queue.");
}
- MockRM mockRM = setupResourceManager();
+ ReservationId reservationID =
+ ReservationId.parseReservationId(reservationId);
- ReservationId reservationID = ReservationId.parseReservationId(reservationId);
- ReservationSystem reservationSystem = mockRM.getReservationSystem();
- reservationSystem.synchronizePlan(QUEUE_DEDICATED_FULL, true);
+ if (!reservationMap.containsKey(reservationID)) {
+ throw new NotFoundException("reservationId with id: " + reservationId + " not found");
+ }
- // Generate reserved resources
ClientRMService clientService = mockRM.getClientRMService();
- // arrival time from which the resource(s) can be allocated.
- long arrival = Time.now();
-
- // deadline by when the resource(s) must be allocated.
- // The reason for choosing 1.05 is because this gives an integer
- // DURATION * 0.05 = 3000(ms)
- // deadline = arrival + 3000ms
- long deadline = (long) (arrival + 1.05 * DURATION);
-
- // In this test of reserved resources, we will apply for 4 containers (1 core, 1GB memory)
- // arrival = Time.now(), and make sure deadline - arrival > duration,
- // the current setting is greater than 3000ms
- ReservationSubmissionRequest submissionRequest =
- ReservationSystemTestUtil.createSimpleReservationRequest(
- reservationID, NUM_CONTAINERS, arrival, deadline, DURATION);
- clientService.submitReservation(submissionRequest);
-
// listReservations
ReservationListRequest request = ReservationListRequest.newInstance(
- queue, reservationID.toString(), startTime, endTime, includeResourceAllocations);
+ queue, reservationId, startTime, endTime, includeResourceAllocations);
ReservationListResponse resRespInfo = clientService.listReservations(request);
ReservationListInfo resResponse =
new ReservationListInfo(resRespInfo, includeResourceAllocations);
- if (mockRM != null) {
- mockRM.stop();
+ return Response.status(Status.OK).entity(resResponse).build();
+ }
+
+ @Override
+ public Response createNewReservation(HttpServletRequest hsr)
+ throws AuthorizationException, IOException, InterruptedException {
+
+ if (!isRunning) {
+ throw new RuntimeException("RM is stopped");
+ }
+
+ ReservationId resId = ReservationId.newInstance(Time.now(), resCounter.incrementAndGet());
+ LOG.info("Allocated new reservationId: {}.", resId);
+
+ NewReservation reservationId = new NewReservation(resId.toString());
+ return Response.status(Status.OK).entity(reservationId).build();
+ }
+
+ @Override
+ public Response submitReservation(ReservationSubmissionRequestInfo resContext,
+ HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException {
+
+ if (!isRunning) {
+ throw new RuntimeException("RM is stopped");
}
- return Response.status(Status.OK).entity(resResponse).build();
+ ReservationId reservationId = ReservationId.parseReservationId(resContext.getReservationId());
+ ReservationDefinitionInfo definitionInfo = resContext.getReservationDefinition();
+ ReservationDefinition definition =
+ RouterServerUtil.convertReservationDefinition(definitionInfo);
+ ReservationSubmissionRequest request = ReservationSubmissionRequest.newInstance(
+ definition, resContext.getQueue(), reservationId);
+ submitReservation(request);
+
+ LOG.info("Reservation submitted: {}.", reservationId);
+
+ SubClusterId subClusterId = getSubClusterId();
+ reservationMap.put(reservationId, subClusterId);
+
+ return Response.status(Status.ACCEPTED).build();
+ }
+
+ private void submitReservation(ReservationSubmissionRequest request) {
+ try {
+ // synchronize plan
+ ReservationSystem reservationSystem = mockRM.getReservationSystem();
+ reservationSystem.synchronizePlan(QUEUE_DEDICATED_FULL, true);
+ // Generate reserved resources
+ ClientRMService clientService = mockRM.getClientRMService();
+ clientService.submitReservation(request);
+ } catch (IOException | YarnException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Response updateReservation(ReservationUpdateRequestInfo resContext,
+ HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException {
+
+ if (resContext == null || resContext.getReservationId() == null ||
+ resContext.getReservationDefinition() == null) {
+ return Response.status(Status.BAD_REQUEST).build();
+ }
+
+ String resId = resContext.getReservationId();
+ ReservationId reservationId = ReservationId.parseReservationId(resId);
+
+ if (!reservationMap.containsKey(reservationId)) {
+ throw new NotFoundException("reservationId with id: " + reservationId + " not found");
+ }
+
+ // Generate reserved resources
+ updateReservation(resContext);
+
+ ReservationUpdateResponseInfo resRespInfo = new ReservationUpdateResponseInfo();
+ return Response.status(Status.OK).entity(resRespInfo).build();
+ }
+
+ private void updateReservation(ReservationUpdateRequestInfo resContext) throws IOException {
+
+ if (resContext == null) {
+ throw new BadRequestException("Input ReservationSubmissionContext should not be null");
+ }
+
+ ReservationDefinitionInfo resInfo = resContext.getReservationDefinition();
+ if (resInfo == null) {
+ throw new BadRequestException("Input ReservationDefinition should not be null");
+ }
+
+ ReservationRequestsInfo resReqsInfo = resInfo.getReservationRequests();
+ if (resReqsInfo == null || resReqsInfo.getReservationRequest() == null
+ || resReqsInfo.getReservationRequest().isEmpty()) {
+ throw new BadRequestException("The ReservationDefinition should " +
+ "contain at least one ReservationRequest");
+ }
+
+ if (resContext.getReservationId() == null) {
+ throw new BadRequestException("Update operations must specify an existing ReservaitonId");
+ }
+
+ ReservationRequestInterpreter[] values = ReservationRequestInterpreter.values();
+ ReservationRequestInterpreter requestInterpreter =
+ values[resReqsInfo.getReservationRequestsInterpreter()];
+ List<ReservationRequest> list = new ArrayList<>();
+
+ for (ReservationRequestInfo resReqInfo : resReqsInfo.getReservationRequest()) {
+ ResourceInfo rInfo = resReqInfo.getCapability();
+ Resource capability = Resource.newInstance(rInfo.getMemorySize(), rInfo.getvCores());
+ int numContainers = resReqInfo.getNumContainers();
+ int minConcurrency = resReqInfo.getMinConcurrency();
+ long duration = resReqInfo.getDuration();
+ ReservationRequest rr = ReservationRequest.newInstance(
+ capability, numContainers, minConcurrency, duration);
+ list.add(rr);
+ }
+
+ ReservationRequests reqs = ReservationRequests.newInstance(list, requestInterpreter);
+ ReservationDefinition rDef = ReservationDefinition.newInstance(
+ resInfo.getArrival(), resInfo.getDeadline(), reqs,
+ resInfo.getReservationName(), resInfo.getRecurrenceExpression(),
+ Priority.newInstance(resInfo.getPriority()));
+ ReservationUpdateRequest request = ReservationUpdateRequest.newInstance(
+ rDef, ReservationId.parseReservationId(resContext.getReservationId()));
+
+ ClientRMService clientService = mockRM.getClientRMService();
+ try {
+ clientService.updateReservation(request);
+ } catch (YarnException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ public Response deleteReservation(ReservationDeleteRequestInfo resContext, HttpServletRequest hsr)
+ throws AuthorizationException, IOException, InterruptedException {
+ if (!isRunning) {
+ throw new RuntimeException("RM is stopped");
+ }
+
+ try {
+ String resId = resContext.getReservationId();
+ ReservationId reservationId = ReservationId.parseReservationId(resId);
+
+ if (!reservationMap.containsKey(reservationId)) {
+ throw new NotFoundException("reservationId with id: " + reservationId + " not found");
+ }
+
+ ReservationDeleteRequest reservationDeleteRequest =
+ ReservationDeleteRequest.newInstance(reservationId);
+ ClientRMService clientService = mockRM.getClientRMService();
+ clientService.deleteReservation(reservationDeleteRequest);
+
+ ReservationDeleteResponseInfo resRespInfo = new ReservationDeleteResponseInfo();
+ reservationMap.remove(reservationId);
+
+ return Response.status(Status.OK).entity(resRespInfo).build();
+ } catch (YarnException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @VisibleForTesting
+ public MockRM getMockRM() {
+ return mockRM;
+ }
+
+ @VisibleForTesting
+ public void setMockRM(MockRM mockRM) {
+ this.mockRM = mockRM;
Review Comment:
Make the param mockResourceManager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org