slfan1989 commented on code in PR #4764:
URL: https://github.com/apache/hadoop/pull/4764#discussion_r953258995
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java:
##########
@@ -888,13 +890,120 @@ public MoveApplicationAcrossQueuesResponse
moveApplicationAcrossQueues(
@Override
public GetNewReservationResponse getNewReservation(
GetNewReservationRequest request) throws YarnException, IOException {
- throw new NotImplementedException("Code is not implemented");
+
+ if (request == null) {
+ routerMetrics.incrGetNewReservationFailedRetrieved();
+ String errMsg = "Missing getNewReservation request.";
+ RouterServerUtil.logAndThrowException(errMsg, null);
+ }
+
+ long startTime = clock.getTime();
+ Map<SubClusterId, SubClusterInfo> subClustersActive =
+ federationFacade.getSubClusters(true);
+
+ for (int i = 0; i < numSubmitRetries; ++i) {
+ SubClusterId subClusterId = getRandomActiveSubCluster(subClustersActive);
+ LOG.info("getNewReservation try #{} on SubCluster {}.", i, subClusterId);
+ ApplicationClientProtocol clientRMProxy =
getClientRMProxyForSubCluster(subClusterId);
+ GetNewReservationResponse response = null;
+ try {
+ response = clientRMProxy.getNewReservation(request);
+ } catch (Exception e) {
+ LOG.warn("Unable to create a new Reservation in SubCluster {}.",
subClusterId.getId(), e);
+ subClustersActive.remove(subClusterId);
+ }
+
+ if (response != null) {
+ long stopTime = clock.getTime();
+ routerMetrics.succeededGetNewReservationRetrieved(stopTime -
startTime);
+ return response;
+ }
+ }
+
+ routerMetrics.incrGetNewReservationFailedRetrieved();
+ String errMsg = "Failed to create a new reservation.";
+ throw new YarnException(errMsg);
}
@Override
public ReservationSubmissionResponse submitReservation(
ReservationSubmissionRequest request) throws YarnException, IOException {
- throw new NotImplementedException("Code is not implemented");
+
+ if (request == null || request.getReservationId() == null
+ || request.getReservationDefinition() == null ||
request.getQueue() == null) {
+ routerMetrics.incrSubmitReservationFailedRetrieved();
+ RouterServerUtil.logAndThrowException(
+ "Missing submitReservation request or reservationId " +
+ "or reservation definition or queue.", null);
+ }
+
+ ReservationSubmissionResponse response = null;
+ long startTime = clock.getTime();
+ ReservationId reservationId = request.getReservationId();
+
+ for (int i = 0; i < numSubmitRetries; ++i) {
+
+ SubClusterId subClusterId =
policyFacade.getReservationHomeSubCluster(request);
+
+ LOG.info("submitReservation reservationId {} try #{} on SubCluster {}.",
+ reservationId, i, subClusterId);
+
+ ReservationHomeSubCluster reservationHomeSubCluster =
+ ReservationHomeSubCluster.newInstance(reservationId, subClusterId);
+
+ if (i == 0) {
Review Comment:
I tried to refactor this part of the code, please help to review the code
again.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
