diff --git hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ResourceMgrDelegate.java hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ResourceMgrDelegate.java index 21c0d0f..af18480 100644 --- hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ResourceMgrDelegate.java +++ hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ResourceMgrDelegate.java @@ -44,6 +44,7 @@ import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.token.Token; import org.apache.hadoop.yarn.api.ApplicationClientProtocol; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteRequest; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationListRequest; @@ -436,6 +437,12 @@ public void moveApplicationAcrossQueues(ApplicationId appId, String queue) } @Override + public GetNewReservationResponse createReservation() throws YarnException, + IOException { + return client.createReservation(); + } + + @Override public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException { return client.submitReservation(request); diff --git hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java index 64f967d..85d89ae 100644 --- hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java +++ hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java @@ -96,6 +96,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -424,6 +426,12 @@ public GetContainersResponse getContainers(GetContainersRequest request) } @Override + public GetNewReservationResponse getNewReservation( + GetNewReservationRequest request) throws YarnException, IOException { + return null; + } + + @Override public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException { return null; diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java index bca062e..d9651b8 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java @@ -37,6 +37,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -302,6 +304,28 @@ public MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues( MoveApplicationAcrossQueuesRequest request) throws YarnException, IOException; /** + *
The interface used by clients to obtain a new {@link ReservationId} for + * submitting new reservations.
+ * + *The ResourceManager responds with a new, monotonically
+ * increasing, {@link ReservationId} which is used by the client to submit
+ * a new reservation.
ReservationId
+ * @return response containing the new ReservationId to be used
+ * to submit an reservation
+ * @throws YarnException if the reservation system is not enabled.
+ * @throws IOException on IO failures.
+ * @see #submitReservation(ReservationSubmissionRequest)
+ */
+ @Public
+ @Stable
+ @Idempotent
+ GetNewReservationResponse getNewReservation(
+ GetNewReservationRequest request)
+ throws YarnException, IOException;
+
+ /**
* * The interface used by clients to submit a new reservation to the * {@code ResourceManager}. @@ -349,6 +373,7 @@ public MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues( */ @Public @Unstable + @Idempotent public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException; diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationRequest.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationRequest.java new file mode 100644 index 0000000..c8ef0ea --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationRequest.java @@ -0,0 +1,41 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.api.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Stable; +import org.apache.hadoop.yarn.util.Records; + +/** + *
The request sent by clients to get a new {@code ReservationId} for + * submitting an reservation.
+ * + * {@code ApplicationClientProtocol#getNewReservation(GetNewReservationRequest)} + */ +@Public +@Stable +public abstract class GetNewReservationRequest { + @Public + @Stable + public static GetNewReservationRequest newInstance() { + GetNewReservationRequest request = + Records.newRecord(GetNewReservationRequest.class); + return request; + } +} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationResponse.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationResponse.java new file mode 100644 index 0000000..6f319f3 --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetNewReservationResponse.java @@ -0,0 +1,71 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.api.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Stable; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.records.ReservationId; +import org.apache.hadoop.yarn.util.Records; + +/** + *The response sent by the ResourceManager to the client for
+ * a request to get a new {@link ReservationId} for submitting reservations.
Clients can submit an reservation with the returned + * {@link ReservationId}.
+ * + * {@code ApplicationClientProtocol#getNewReservation(GetNewReservationRequest)} + */ +@Public +@Stable +public abstract class GetNewReservationResponse { + + @Private + @Unstable + public static GetNewReservationResponse newInstance( + ReservationId reservationId) { + GetNewReservationResponse response = + Records.newRecord(GetNewReservationResponse.class); + response.setReservationId(reservationId); + return response; + } + + /** + * Get a new {@link ReservationId} to be used to submit a reservation. + * + * @return a {@link ReservationId} representing the unique id to identify + * a reservation with which it was submitted. + */ + @Public + @Unstable + public abstract ReservationId getReservationId(); + + /** + * Set a new {@link ReservationId} to be used to submit a reservation. + * + * @param reservationId a {@link ReservationId} representing the unique id to + * identify a reservation with which it was submitted. + */ + @Private + @Unstable + public abstract void setReservationId(ReservationId reservationId); + +} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionRequest.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionRequest.java index 6d5ca16..3872f41 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionRequest.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionRequest.java @@ -22,6 +22,7 @@ import org.apache.hadoop.classification.InterfaceStability.Unstable; import org.apache.hadoop.yarn.api.records.QueueInfo; import org.apache.hadoop.yarn.api.records.ReservationDefinition; +import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.util.Records; /** @@ -38,11 +39,13 @@ @Public @Unstable public static ReservationSubmissionRequest newInstance( - ReservationDefinition reservationDefinition, String queueName) { + ReservationDefinition reservationDefinition, String queueName, + ReservationId reservationId) { ReservationSubmissionRequest request = Records.newRecord(ReservationSubmissionRequest.class); request.setReservationDefinition(reservationDefinition); request.setQueue(queueName); + request.setReservationId(reservationId); return request; } @@ -94,4 +97,24 @@ public abstract void setReservationDefinition( @Unstable public abstract void setQueue(String queueName); + /** + * Get the reservation id that corresponds to the reservation submission. + * + * @return reservation id that will be used to identify the reservation + * submission. + */ + @Public + @Unstable + public abstract ReservationId getReservationId(); + + /** + * Set the reservation id that corresponds to the reservation submission. + * + * @param reservationId reservation id that will be used to identify the + * reservation submission. + */ + @Public + @Unstable + public abstract void setReservationId(ReservationId reservationId); + } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionResponse.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionResponse.java index 32fe5e0..bf5b97f 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionResponse.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ReservationSubmissionResponse.java @@ -22,13 +22,12 @@ import org.apache.hadoop.classification.InterfaceAudience.Public; import org.apache.hadoop.classification.InterfaceStability.Unstable; import org.apache.hadoop.yarn.api.records.ReservationDefinition; -import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.util.Records; /** * {@link ReservationSubmissionResponse} contains the answer of the admission * control system in the {@code ResourceManager} to a reservation create - * operation. Response contains a {@link ReservationId} if the operation was + * operation. Response contains a {@code ReservationId} if the operation was * successful, if not an exception reporting reason for a failure. * * @see ReservationDefinition @@ -40,37 +39,10 @@ @Private @Unstable - public static ReservationSubmissionResponse newInstance( - ReservationId reservationId) { + public static ReservationSubmissionResponse newInstance() { ReservationSubmissionResponse response = Records.newRecord(ReservationSubmissionResponse.class); - response.setReservationId(reservationId); return response; } - /** - * Get the {@link ReservationId}, that corresponds to a valid resource - * allocation in the scheduler (between start and end time of this - * reservation) - * - * @return the {@link ReservationId} representing the unique id of the - * corresponding reserved resource allocation in the scheduler - */ - @Public - @Unstable - public abstract ReservationId getReservationId(); - - /** - * Set the {@link ReservationId}, that correspond to a valid resource - * allocation in the scheduler (between start and end time of this - * reservation) - * - * @param reservationId the {@link ReservationId} representing the the unique - * id of the corresponding reserved resource allocation in the - * scheduler - */ - @Private - @Unstable - public abstract void setReservationId(ReservationId reservationId); - } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto index 763c839..7046b24 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto @@ -50,6 +50,7 @@ service ApplicationClientProtocolService { rpc getApplicationAttempts (GetApplicationAttemptsRequestProto) returns (GetApplicationAttemptsResponseProto); rpc getContainerReport (GetContainerReportRequestProto) returns (GetContainerReportResponseProto); rpc getContainers (GetContainersRequestProto) returns (GetContainersResponseProto); + rpc getNewReservation (GetNewReservationRequestProto) returns (GetNewReservationResponseProto); rpc submitReservation (ReservationSubmissionRequestProto) returns (ReservationSubmissionResponseProto); rpc updateReservation (ReservationUpdateRequestProto) returns (ReservationUpdateResponseProto); rpc deleteReservation (ReservationDeleteRequestProto) returns (ReservationDeleteResponseProto); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto index bdf022f..7070e38 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto @@ -367,13 +367,20 @@ message ReleaseSharedCacheResourceResponseProto { // reservation_protocol ////////////////////////////////////////////////////// +message GetNewReservationRequestProto { +} + +message GetNewReservationResponseProto { + optional ReservationIdProto reservation_id = 1; +} + message ReservationSubmissionRequestProto { optional string queue = 1; optional ReservationDefinitionProto reservation_definition = 2; + optional ReservationIdProto reservation_id = 3; } message ReservationSubmissionResponseProto { - optional ReservationIdProto reservation_id = 1; } message ReservationUpdateRequestProto { diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java index ff231a8..d4df20e 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java @@ -32,6 +32,7 @@ import org.apache.hadoop.io.Text; import org.apache.hadoop.service.AbstractService; import org.apache.hadoop.yarn.api.ApplicationClientProtocol; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteRequest; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationListRequest; @@ -535,6 +536,20 @@ public abstract void moveApplicationAcrossQueues(ApplicationId appId, /** *+ * Obtain a {@link GetNewReservationResponse} for a new reservation, + * which contains the {@link ReservationId} object. + *
+ * + * @return The {@link GetNewReservationResponse} containing a new + * {@link ReservationId} object. + * @throws YarnException + * @throws IOException + */ + public abstract GetNewReservationResponse createReservation() + throws YarnException, IOException; + + /** + ** The interface used by clients to submit a new reservation to the * {@code ResourceManager}. *
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java index 56e42c4..36e6dfe 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java @@ -65,6 +65,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoRequest; @@ -789,6 +791,14 @@ public void moveApplicationAcrossQueues(ApplicationId appId, } @Override + public GetNewReservationResponse createReservation() throws YarnException, + IOException { + GetNewReservationRequest request = + Records.newRecord(GetNewReservationRequest.class); + return rmClient.getNewReservation(request); + } + + @Override public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException { return rmClient.submitReservation(request); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java index 6e0930c..7d763cd 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java @@ -67,6 +67,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetContainersResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest; @@ -988,7 +990,19 @@ private ApplicationId createApp(YarnClient rmClient, boolean unmanaged) return appId; } - + + private GetNewReservationResponse getNewReservation(YarnClient rmClient) { + GetNewReservationRequest newReservationRequest = GetNewReservationRequest + .newInstance(); + GetNewReservationResponse getNewReservationResponse = null; + try { + getNewReservationResponse = rmClient.createReservation(); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + return getNewReservationResponse; + } + private void waitTillAccepted(YarnClient rmClient, ApplicationId appId, boolean unmanagedApplication) throws Exception { @@ -1202,7 +1216,7 @@ public void testReservationAPIs() { for(attempts = 10; attempts > 0; attempts--) { if (cluster.getResourceManager().getRMContext().getReservationSystem() .getPlan(ReservationSystemTestUtil.reservationQ).getTotalCapacity() - .getMemory() > 0) { + .getMemory() > 6000) { break; } try { @@ -1221,8 +1235,11 @@ public void testReservationAPIs() { long arrival = clock.getTime(); long duration = 60000; long deadline = (long) (arrival + 1.05 * duration); + ReservationId reservationID = getNewReservation(client) + .getReservationId(); ReservationSubmissionRequest sRequest = - createSimpleReservationRequest(4, arrival, deadline, duration); + createSimpleReservationRequest(reservationID, 4, arrival, deadline, + duration); ReservationSubmissionResponse sResponse = null; try { sResponse = client.submitReservation(sRequest); @@ -1230,12 +1247,33 @@ public void testReservationAPIs() { Assert.fail(e.getMessage()); } Assert.assertNotNull(sResponse); - ReservationId reservationID = sResponse.getReservationId(); Assert.assertNotNull(reservationID); System.out.println("Submit reservation response: " + reservationID); - // Update the reservation + // Submit the reservation again with the same request and make sure it + // passes. + try { + sResponse = client.submitReservation(sRequest); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + + // Submit the reservation with the same reservation id but different + // reservation definition, and ensure YarnException is thrown. + arrival = clock.getTime(); ReservationDefinition rDef = sRequest.getReservationDefinition(); + rDef.setArrival(arrival + duration); + sRequest.setReservationDefinition(rDef); + try { + sResponse = client.submitReservation(sRequest); + Assert.fail("Reservation submission should fail if a duplicate " + + "reservation id is used, but the reservation definition has been " + + "updated."); + } catch (Exception e) { + Assert.assertTrue(e instanceof YarnException); + } + + // Update the reservation ReservationRequest rr = rDef.getReservationRequests().getReservationResources().get(0); rr.setNumContainers(5); @@ -1448,7 +1486,8 @@ public void testReservationAPIs() { } private ReservationSubmissionRequest createSimpleReservationRequest( - int numContainers, long arrival, long deadline, long duration) { + ReservationId reservationId, int numContainers, long arrival, + long deadline, long duration) { // create a request with a single atomic ask ReservationRequest r = ReservationRequest.newInstance(Resource.newInstance(1024, 1), @@ -1461,7 +1500,7 @@ private ReservationSubmissionRequest createSimpleReservationRequest( "testYarnClient#reservation"); ReservationSubmissionRequest request = ReservationSubmissionRequest.newInstance(rDef, - ReservationSystemTestUtil.reservationQ); + ReservationSystemTestUtil.reservationQ, reservationId); return request; } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java index e5aad74..4d65425 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java @@ -59,6 +59,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -113,6 +115,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetLabelsToNodesResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewApplicationRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewApplicationResponsePBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewReservationRequestPBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewReservationResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNodesToLabelsRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNodesToLabelsResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetQueueInfoRequestPBImpl; @@ -453,6 +457,21 @@ public GetContainersResponse getContainers(GetContainersRequest request) } @Override + public GetNewReservationResponse getNewReservation( + GetNewReservationRequest request) + throws YarnException, IOException { + YarnServiceProtos.GetNewReservationRequestProto requestProto = + ((GetNewReservationRequestPBImpl) request).getProto(); + try { + return new GetNewReservationResponsePBImpl(proxy.getNewReservation(null, + requestProto)); + } catch (ServiceException e) { + RPCUtil.unwrapAndThrowException(e); + return null; + } + } + + @Override public ReservationSubmissionResponse submitReservation(ReservationSubmissionRequest request) throws YarnException, IOException { ReservationSubmissionRequestProto requestProto = diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java index 2c5794e..4923794 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java @@ -43,6 +43,7 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoResponse; @@ -84,6 +85,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetLabelsToNodesResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewApplicationRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewApplicationResponsePBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewReservationRequestPBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNewReservationResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNodesToLabelsRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetNodesToLabelsResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetQueueInfoRequestPBImpl; @@ -136,6 +139,8 @@ import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetLabelsToNodesResponseProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewApplicationRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewApplicationResponseProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewReservationRequestProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewReservationResponseProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNodesToLabelsRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNodesToLabelsResponseProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetQueueInfoRequestProto; @@ -449,6 +454,22 @@ public GetContainersResponseProto getContainers(RpcController controller, } @Override + public GetNewReservationResponseProto getNewReservation( + RpcController arg0, GetNewReservationRequestProto proto) throws + ServiceException { + GetNewReservationRequestPBImpl request = + new GetNewReservationRequestPBImpl(proto); + try { + GetNewReservationResponse response = real.getNewReservation(request); + return ((GetNewReservationResponsePBImpl)response).getProto(); + } catch (YarnException e) { + throw new ServiceException(e); + } catch (IOException e) { + throw new ServiceException(e); + } + } + + @Override public ReservationSubmissionResponseProto submitReservation(RpcController controller, ReservationSubmissionRequestProto requestProto) throws ServiceException { ReservationSubmissionRequestPBImpl request = diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationRequestPBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationRequestPBImpl.java new file mode 100644 index 0000000..b43f6b5 --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationRequestPBImpl.java @@ -0,0 +1,77 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.api.protocolrecords.impl.pb; + + +import com.google.protobuf.TextFormat; +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewReservationRequestProto; + +/** + *The implementation of the request sent by clients to get a + * new {@code ReservationId} for submitting an reservation.
+ * + * {@code ApplicationClientProtocol#getNewReservation(GetNewReservationRequest)} + */ +@Private +@Unstable +public class GetNewReservationRequestPBImpl extends GetNewReservationRequest { + private GetNewReservationRequestProto proto = + GetNewReservationRequestProto.getDefaultInstance(); + private GetNewReservationRequestProto.Builder builder = null; + private boolean viaProto = false; + + public GetNewReservationRequestPBImpl(GetNewReservationRequestProto proto) { + this.proto = proto; + viaProto = true; + } + + public GetNewReservationRequestPBImpl() { + builder = GetNewReservationRequestProto.newBuilder(); + } + + public GetNewReservationRequestProto getProto() { + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public int hashCode() { + return getProto().hashCode(); + } + + @Override + public boolean equals(Object other) { + if (other == null) { + return false; + } + if (other.getClass().isAssignableFrom(this.getClass())) { + return this.getProto().equals(this.getClass().cast(other).getProto()); + } + return false; + } + + @Override + public String toString() { + return TextFormat.shortDebugString(getProto()); + } +} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationResponsePBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationResponsePBImpl.java new file mode 100644 index 0000000..c09c4e2 --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetNewReservationResponsePBImpl.java @@ -0,0 +1,144 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.api.protocolrecords.impl.pb; + + +import com.google.protobuf.TextFormat; +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; +import org.apache.hadoop.yarn.api.records.ReservationId; +import org.apache.hadoop.yarn.api.records.impl.pb.ReservationIdPBImpl; +import org.apache.hadoop.yarn.proto.YarnProtos.ReservationIdProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewReservationResponseProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.GetNewReservationResponseProtoOrBuilder; + + +/** + *The implementation of the response sent by the
+ * ResourceManager to the client for a request to get a new
+ * {@link ReservationId} for submitting reservations.
Clients can submit an reservation with the returned + * {@link ReservationId}.
+ * + * {@code ApplicationClientProtocol#getNewReservation(GetNewReservationRequest)} + */ +@Private +@Unstable +public class GetNewReservationResponsePBImpl extends GetNewReservationResponse { + private GetNewReservationResponseProto proto = + GetNewReservationResponseProto.getDefaultInstance(); + private GetNewReservationResponseProto.Builder builder = null; + private boolean viaProto = false; + + private ReservationId reservationId = null; + + public GetNewReservationResponsePBImpl() { + builder = GetNewReservationResponseProto.newBuilder(); + } + + public GetNewReservationResponsePBImpl(GetNewReservationResponseProto proto) { + this.proto = proto; + viaProto = true; + } + + public GetNewReservationResponseProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public int hashCode() { + return getProto().hashCode(); + } + + @Override + public boolean equals(Object other) { + if (other == null) { + return false; + } + if (other.getClass().isAssignableFrom(this.getClass())) { + return this.getProto().equals(this.getClass().cast(other).getProto()); + } + return false; + } + + @Override + public String toString() { + return TextFormat.shortDebugString(getProto()); + } + + private void mergeLocalToBuilder() { + if (reservationId != null) { + builder.setReservationId(convertToProtoFormat(this.reservationId)); + } + } + + private void mergeLocalToProto() { + if (viaProto) { + maybeInitBuilder(); + } + mergeLocalToBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = GetNewReservationResponseProto.newBuilder(proto); + } + viaProto = false; + } + + @Override + public ReservationId getReservationId() { + if (this.reservationId != null) { + return this.reservationId; + } + + GetNewReservationResponseProtoOrBuilder p = viaProto ? proto : builder; + if (!p.hasReservationId()) { + return null; + } + + this.reservationId = convertFromProtoFormat(p.getReservationId()); + return this.reservationId; + } + + @Override + public void setReservationId(ReservationId reservationId) { + maybeInitBuilder(); + if (reservationId == null) { + builder.clearReservationId(); + } + this.reservationId = reservationId; + } + + private ReservationIdPBImpl convertFromProtoFormat(ReservationIdProto p) { + return new ReservationIdPBImpl(p); + } + + private ReservationIdProto convertToProtoFormat(ReservationId t) { + return ((ReservationIdPBImpl)t).getProto(); + } + +} \ No newline at end of file diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionRequestPBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionRequestPBImpl.java index 9f10f4c..016d4de 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionRequestPBImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionRequestPBImpl.java @@ -20,8 +20,11 @@ import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest; import org.apache.hadoop.yarn.api.records.ReservationDefinition; +import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.api.records.impl.pb.ReservationDefinitionPBImpl; +import org.apache.hadoop.yarn.api.records.impl.pb.ReservationIdPBImpl; import org.apache.hadoop.yarn.proto.YarnProtos.ReservationDefinitionProto; +import org.apache.hadoop.yarn.proto.YarnProtos.ReservationIdProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationSubmissionRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationSubmissionRequestProtoOrBuilder; @@ -119,6 +122,25 @@ public void setQueue(String planName) { builder.setQueue(planName); } + @Override + public ReservationId getReservationId() { + ReservationSubmissionRequestProtoOrBuilder p = viaProto ? proto : builder; + if (!p.hasReservationId()) { + return null; + } + return (convertFromProtoFormat(p.getReservationId())); + } + + @Override + public void setReservationId(ReservationId reservationId) { + maybeInitBuilder(); + if (reservationId == null) { + builder.clearReservationId(); + return; + } + builder.setReservationId(convertToProtoFormat(reservationId)); + } + private ReservationDefinitionProto convertToProtoFormat( ReservationDefinition r) { return ((ReservationDefinitionPBImpl) r).getProto(); @@ -129,6 +151,14 @@ private ReservationDefinitionPBImpl convertFromProtoFormat( return new ReservationDefinitionPBImpl(r); } + private ReservationIdProto convertToProtoFormat(ReservationId r) { + return ((ReservationIdPBImpl) r).getProto(); + } + + private ReservationIdPBImpl convertFromProtoFormat(ReservationIdProto r) { + return new ReservationIdPBImpl(r); + } + @Override public int hashCode() { return getProto().hashCode(); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionResponsePBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionResponsePBImpl.java index 82151f4..81b3a84 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionResponsePBImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ReservationSubmissionResponsePBImpl.java @@ -19,11 +19,7 @@ package org.apache.hadoop.yarn.api.protocolrecords.impl.pb; import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionResponse; -import org.apache.hadoop.yarn.api.records.ReservationId; -import org.apache.hadoop.yarn.api.records.impl.pb.ReservationIdPBImpl; -import org.apache.hadoop.yarn.proto.YarnProtos.ReservationIdProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationSubmissionResponseProto; -import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationSubmissionResponseProtoOrBuilder; import com.google.protobuf.TextFormat; @@ -35,8 +31,6 @@ ReservationSubmissionResponseProto.Builder builder = null; boolean viaProto = false; - private ReservationId reservationId; - public ReservationSubmissionResponsePBImpl() { builder = ReservationSubmissionResponseProto.newBuilder(); } @@ -48,64 +42,11 @@ public ReservationSubmissionResponsePBImpl( } public ReservationSubmissionResponseProto getProto() { - mergeLocalToProto(); proto = viaProto ? proto : builder.build(); viaProto = true; return proto; } - private void mergeLocalToBuilder() { - if (this.reservationId != null) { - builder.setReservationId(convertToProtoFormat(this.reservationId)); - } - } - - private void mergeLocalToProto() { - if (viaProto) - maybeInitBuilder(); - mergeLocalToBuilder(); - proto = builder.build(); - viaProto = true; - } - - private void maybeInitBuilder() { - if (viaProto || builder == null) { - builder = ReservationSubmissionResponseProto.newBuilder(proto); - } - viaProto = false; - } - - @Override - public ReservationId getReservationId() { - ReservationSubmissionResponseProtoOrBuilder p = viaProto ? proto : builder; - if (reservationId != null) { - return reservationId; - } - if (!p.hasReservationId()) { - return null; - } - reservationId = convertFromProtoFormat(p.getReservationId()); - return reservationId; - } - - @Override - public void setReservationId(ReservationId reservationId) { - maybeInitBuilder(); - if (reservationId == null) { - builder.clearReservationId(); - return; - } - this.reservationId = reservationId; - } - - private ReservationIdPBImpl convertFromProtoFormat(ReservationIdProto p) { - return new ReservationIdPBImpl(p); - } - - private ReservationIdProto convertToProtoFormat(ReservationId t) { - return ((ReservationIdPBImpl) t).getProto(); - } - @Override public int hashCode() { return getProto().hashCode(); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java index c512f8d..1eb253e 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java @@ -61,6 +61,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -426,6 +428,12 @@ public GetContainersResponse getContainers(GetContainersRequest request) } @Override + public GetNewReservationResponse getNewReservation( + GetNewReservationRequest request) throws YarnException, IOException { + throw new NotImplementedException(); + } + + @Override public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException { diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java index d68b7b1..6cb3149 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java @@ -79,6 +79,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -1218,18 +1220,46 @@ public Server getServer() { } @Override + public GetNewReservationResponse getNewReservation( + GetNewReservationRequest request) throws YarnException, IOException { + checkReservationSytem(AuditConstants.CREATE_NEW_RESERVATION_REQUEST); + GetNewReservationResponse response = + recordFactory.newRecordInstance(GetNewReservationResponse.class); + + ReservationId reservationId = reservationSystem.getNewReservationId(); + response.setReservationId(reservationId); + // Create a new Reservation Id + return response; + } + + @Override public ReservationSubmissionResponse submitReservation( ReservationSubmissionRequest request) throws YarnException, IOException { // Check if reservation system is enabled checkReservationSytem(AuditConstants.SUBMIT_RESERVATION_REQUEST); ReservationSubmissionResponse response = recordFactory.newRecordInstance(ReservationSubmissionResponse.class); - // Create a new Reservation Id - ReservationId reservationId = reservationSystem.getNewReservationId(); + ReservationId reservationId = request.getReservationId(); // Validate the input Plan plan = rValidator.validateReservationSubmissionRequest(reservationSystem, request, reservationId); + + ReservationAllocation allocation = plan.getReservationById(reservationId); + + if (allocation != null) { + boolean isNewDefinition = !allocation.getReservationDefinition().equals( + request.getReservationDefinition()); + if (isNewDefinition) { + String message = "Reservation allocation already exists with the " + + "reservation id " + reservationId.toString() + ". Please try " + + "again with a new reservation id."; + throw RPCUtil.getRemoteException(message); + } else { + return response; + } + } + // Check ACLs String queueName = request.getQueue(); String user = @@ -1248,7 +1278,6 @@ public ReservationSubmissionResponse submitReservation( refreshScheduler(queueName, request.getReservationDefinition(), reservationId.toString()); // return the reservation id - response.setReservationId(reservationId); } } catch (PlanningException e) { RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_RESERVATION_REQUEST, diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java index d08cb9e..c5bf000 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java @@ -69,6 +69,8 @@ public static final String UNAUTHORIZED_USER = "Unauthorized user"; // For Reservation system + public static final String CREATE_NEW_RESERVATION_REQUEST = "Create " + + "Reservation Request"; public static final String SUBMIT_RESERVATION_REQUEST = "Submit Reservation Request"; public static final String UPDATE_RESERVATION_REQUEST = "Update Reservation Request"; public static final String DELETE_RESERVATION_REQUEST = "Delete Reservation Request"; diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java index d63e725..8ebde0a 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java @@ -206,10 +206,17 @@ public Plan validateReservationSubmissionRequest( ReservationSystem reservationSystem, ReservationSubmissionRequest request, ReservationId reservationId) throws YarnException { + String message; + if (reservationId == null) { + message = "Reservation id cannot be null. Please try again " + + "specifying a valid reservation id by creating a new reservation id."; + throw RPCUtil.getRemoteException(message); + } // Check if it is a managed queue String queue = request.getQueue(); Plan plan = getPlanFromQueue(reservationSystem, queue, AuditConstants.SUBMIT_RESERVATION_REQUEST); + validateReservationDefinition(reservationId, request.getReservationDefinition(), plan, AuditConstants.SUBMIT_RESERVATION_REQUEST); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java index 50c450b..d53f952 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java @@ -78,6 +78,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest; import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse; import org.apache.hadoop.yarn.api.protocolrecords.MoveApplicationAcrossQueuesRequest; @@ -149,6 +151,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.LocalResourceInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.LogAggregationContextInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NewApplication; +import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NewReservation; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeLabelInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeLabelsInfo; @@ -162,7 +165,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationListInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationRequestInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationRequestsInfo; -import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationSubmissionResponseInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationSubmissionRequestInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationUpdateRequestInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationUpdateResponseInfo; @@ -1870,6 +1872,59 @@ public CancelDelegationTokenResponse run() throws IOException, } /** + * Generates a new ReservationId which is then sent to the client. + * + * @param hsr the servlet request + * @return Response containing the app id and the maximum resource + * capabilities + * @throws AuthorizationException + * @throws IOException + * @throws InterruptedException + */ + @POST + @Path("/reservation/new-reservation") + @Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML }) + public Response createNewReservation(@Context HttpServletRequest hsr) + throws AuthorizationException, IOException, InterruptedException { + init(); + UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true); + if (callerUGI == null) { + throw new AuthorizationException("Unable to obtain user name, " + + "user not authenticated"); + } + if (UserGroupInformation.isSecurityEnabled() && isStaticUser(callerUGI)) { + String msg = "The default static user cannot carry out this operation."; + return Response.status(Status.FORBIDDEN).entity(msg).build(); + } + + NewReservation reservationId = createNewReservation(); + return Response.status(Status.OK).entity(reservationId).build(); + + } + + /** + * Function that actually creates the {@link ReservationId} by calling the + * ClientRMService. + * + * @return returns structure containing the {@link ReservationId} + */ + private NewReservation createNewReservation() throws IOException { + GetNewReservationRequest req = + recordFactory.newRecordInstance(GetNewReservationRequest.class); + GetNewReservationResponse resp; + try { + resp = rm.getClientRMService().getNewReservation(req); + } catch (YarnException e) { + String msg = "Unable to create new reservation from RM web service"; + LOG.error(msg, e); + throw new YarnRuntimeException(msg, e); + } + NewReservation reservationId = + new NewReservation(resp.getReservationId().toString()); + return reservationId; + } + + /** * Function to submit a Reservation to the RM. * * @param resContext provides information to construct the @@ -1903,19 +1958,15 @@ public Response submitReservation( final ReservationSubmissionRequest reservation = createReservationSubmissionRequest(resContext); - ReservationSubmissionResponseInfo resRespInfo; try { - resRespInfo = - callerUGI.doAs( - new PrivilegedExceptionActionThe response sent by the ResourceManager to the client for
+ * a request to get a new {@code ReservationId} for submitting reservations
+ * using the REST API.
Clients can submit a reservation with the returned {@code ReservationId}. + *
+ * + * {@code RMWebServices#createNewReservation(HttpServletRequest)} + */ +@XmlRootElement(name="new-reservation") +@XmlAccessorType(XmlAccessType.FIELD) +public class NewReservation { + + @XmlElement(name="reservation-id") + private String reservationId; + + public NewReservation() { + reservationId = ""; + } + + public NewReservation(String resId) { + reservationId = resId; + } + + public String getReservationId() { + return reservationId; + } + +} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionRequestInfo.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionRequestInfo.java index 701370d..5ca7998 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionRequestInfo.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionRequestInfo.java @@ -37,6 +37,9 @@ @XmlElement(name = "reservation-definition") private ReservationDefinitionInfo reservationDefinition; + @XmlElement(name = "reservation-id") + private String reservationId; + public ReservationSubmissionRequestInfo() { } @@ -48,6 +51,14 @@ public void setQueue(String queue) { this.queue = queue; } + public String getReservationId() { + return reservationId; + } + + public void setReservationId(String reservationId) { + this.reservationId = reservationId; + } + public ReservationDefinitionInfo getReservationDefinition() { return reservationDefinition; } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionResponseInfo.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionResponseInfo.java deleted file mode 100644 index 943390b..0000000 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationSubmissionResponseInfo.java +++ /dev/null @@ -1,54 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.hadoop.yarn.server.resourcemanager.webapp.dao; - -import javax.xml.bind.annotation.XmlAccessType; -import javax.xml.bind.annotation.XmlAccessorType; -import javax.xml.bind.annotation.XmlElement; -import javax.xml.bind.annotation.XmlRootElement; - -import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionResponse; - -/** - * Simple class that represent a response to a reservation submission. - */ -@XmlRootElement(name = "reservation-submission-response") -@XmlAccessorType(XmlAccessType.FIELD) -public class ReservationSubmissionResponseInfo { - - @XmlElement(name = "reservation-id") - private String reservationId; - - public ReservationSubmissionResponseInfo() { - - } - - public ReservationSubmissionResponseInfo( - ReservationSubmissionResponse response) { - this.reservationId = response.getReservationId().toString(); - } - - public String getReservationId() { - return reservationId; - } - - public void setReservationId(String reservationId) { - this.reservationId = reservationId; - } -} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ReservationACLsTestBase.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ReservationACLsTestBase.java index 4039f50..e15def2 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ReservationACLsTestBase.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ReservationACLsTestBase.java @@ -32,6 +32,8 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.security.authorize.AccessControlList; import org.apache.hadoop.yarn.api.ApplicationClientProtocol; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteRequest; import org.apache.hadoop.yarn.api.protocolrecords.ReservationListRequest; import org.apache.hadoop.yarn.api.protocolrecords.ReservationListResponse; @@ -242,8 +244,8 @@ public void testApplicationACLs() throws Exception { private void verifySubmitReservationSuccess(String submitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(submitter, queueName); + ReservationId reservationId = createReservation(submitter); + submitReservation(submitter, queueName, reservationId); deleteReservation(submitter, reservationId); } @@ -251,7 +253,8 @@ private void verifySubmitReservationSuccess(String submitter, String private void verifySubmitReservationFailure(String submitter, String queueName) throws Exception { try { - submitReservation(submitter, queueName); + ReservationId reservationId = createReservation(submitter); + submitReservation(submitter, queueName, reservationId); Assert.fail("Submit reservation by the enemy should fail!"); } catch (YarnException e) { handleAdministerException(e, submitter, queueName, ReservationACL @@ -261,8 +264,8 @@ private void verifySubmitReservationFailure(String submitter, String private void verifyListReservationSuccess(String lister, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); ReservationListResponse adminResponse = listReservation(lister, queueName); @@ -275,8 +278,8 @@ private void verifyListReservationSuccess(String lister, String private void verifyListReservationFailure(String lister, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); try { listReservation(lister, queueName); @@ -291,8 +294,8 @@ private void verifyListReservationFailure(String lister, private void verifyListReservationByIdSuccess(String lister, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); ReservationListResponse adminResponse = listReservationById(lister, reservationId, queueName); @@ -306,8 +309,8 @@ private void verifyListReservationByIdSuccess(String lister, String private void verifyListReservationByIdFailure(String lister, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); try { listReservationById(lister, reservationId, queueName); Assert.fail("List reservation by the enemy should fail!"); @@ -321,8 +324,8 @@ private void verifyListReservationByIdFailure(String lister, private void verifyDeleteReservationSuccess(String killer, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); deleteReservation(killer, reservationId); } @@ -330,8 +333,8 @@ private void verifyDeleteReservationSuccess(String killer, private void verifyDeleteReservationFailure(String killer, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); try { deleteReservation(killer, reservationId); @@ -346,8 +349,8 @@ private void verifyDeleteReservationFailure(String killer, private void verifyUpdateReservationSuccess(String updater, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); final ReservationUpdateRequest updateRequest = ReservationUpdateRequest.newInstance( @@ -362,8 +365,8 @@ private void verifyUpdateReservationSuccess(String updater, private void verifyUpdateReservationFailure(String updater, String originalSubmitter, String queueName) throws Exception { - ReservationId reservationId = - submitReservation(originalSubmitter, queueName); + ReservationId reservationId = createReservation(originalSubmitter); + submitReservation(originalSubmitter, queueName, reservationId); final ReservationUpdateRequest updateRequest = ReservationUpdateRequest.newInstance( @@ -422,17 +425,27 @@ private void deleteReservation(String deleter, ReservationId id) throws deleteClient.deleteReservation(deleteRequest); } - private ReservationId submitReservation(String submitter, - String queueName) throws Exception { + private ReservationId createReservation(String creator) throws Exception { + + ApplicationClientProtocol creatorClient = getRMClientForUser(creator); + GetNewReservationRequest getNewReservationRequest = + GetNewReservationRequest.newInstance(); + + GetNewReservationResponse response = creatorClient + .getNewReservation(getNewReservationRequest); + return response.getReservationId(); + } + + private void submitReservation(String submitter, + String queueName, ReservationId reservationId) throws Exception { ApplicationClientProtocol submitterClient = getRMClientForUser(submitter); ReservationSubmissionRequest reservationSubmissionRequest = ReservationSubmissionRequest.newInstance( - makeSimpleReservationDefinition(), queueName); + makeSimpleReservationDefinition(), queueName, reservationId); ReservationSubmissionResponse response = submitterClient .submitReservation(reservationSubmissionRequest); - return response.getReservationId(); } private void handleAdministerException(Exception e, String user, String diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java index 8a6ddae..1781340 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java @@ -68,6 +68,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetContainersResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetLabelsToNodesResponse; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsRequest; import org.apache.hadoop.yarn.api.protocolrecords.GetNodesToLabelsResponse; import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest; @@ -139,6 +141,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager; +import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NewReservation; import org.apache.hadoop.yarn.server.security.ApplicationACLsManager; import org.apache.hadoop.yarn.server.utils.BuilderUtils; import org.apache.hadoop.yarn.util.Clock; @@ -1118,9 +1121,18 @@ public void testReservationAPIs() { long arrival = clock.getTime(); long duration = 60000; long deadline = (long) (arrival + 1.05 * duration); + GetNewReservationRequest newReservationRequest = GetNewReservationRequest + .newInstance(); + ReservationId reservationID = null; + try { + reservationID = clientService.getNewReservation( + newReservationRequest).getReservationId(); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } ReservationSubmissionRequest sRequest = - ReservationSystemTestUtil.createSimpleReservationRequest(4, arrival, - deadline, duration); + ReservationSystemTestUtil.createSimpleReservationRequest( + reservationID, 4, arrival, deadline, duration); ReservationSubmissionResponse sResponse = null; try { sResponse = clientService.submitReservation(sRequest); @@ -1128,12 +1140,34 @@ public void testReservationAPIs() { Assert.fail(e.getMessage()); } Assert.assertNotNull(sResponse); - ReservationId reservationID = sResponse.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); - // Update the reservation + // Submit the reservation again with the same request and make sure it + // passes. + try { + sResponse = clientService.submitReservation(sRequest); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + + // Submit the reservation with the same reservation id but different + // reservation definition, and ensure YarnException is thrown. + arrival = clock.getTime(); ReservationDefinition rDef = sRequest.getReservationDefinition(); + rDef.setArrival(arrival + duration); + sRequest.setReservationDefinition(rDef); + try { + sResponse = clientService.submitReservation(sRequest); + Assert.fail("Reservation submission should fail if a duplicate " + + "reservation id is used, but the reservation definition has been " + + "updated."); + } catch (Exception e) { + Assert.assertTrue(e instanceof YarnException); + } + + // Update the reservation + rDef = sRequest.getReservationDefinition(); ReservationRequest rr = rDef.getReservationRequests().getReservationResources().get(0); rr.setNumContainers(5); diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java index 9a0f2c9..5259935 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java @@ -17,6 +17,8 @@ */ package org.apache.hadoop.yarn.server.resourcemanager; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteRequest; import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest; @@ -65,8 +67,11 @@ public void testSubmitReservationAndCheckAfterFailover() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -74,7 +79,6 @@ public void testSubmitReservationAndCheckAfterFailover() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); @@ -98,8 +102,12 @@ public void testUpdateReservationAndCheckAfterFailover() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); + // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -107,7 +115,6 @@ public void testUpdateReservationAndCheckAfterFailover() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); ReservationDefinition reservationDefinition = @@ -144,8 +151,12 @@ public void testDeleteReservationAndCheckAfterFailover() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); + // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -153,7 +164,6 @@ public void testDeleteReservationAndCheckAfterFailover() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); // Delete the reservation @@ -199,13 +209,14 @@ private void addNodeCapacityToPlan(MockRM rm, int memory, int vCores) { } } - private ReservationSubmissionRequest createReservationSubmissionRequest() { + private ReservationSubmissionRequest createReservationSubmissionRequest( + ReservationId reservationId) { Clock clock = new UTCClock(); long arrival = clock.getTime(); long duration = 60000; long deadline = (long) (arrival + duration + 1500); - return ReservationSystemTestUtil.createSimpleReservationRequest(4, arrival, - deadline, duration); + return ReservationSystemTestUtil.createSimpleReservationRequest( + reservationId, 4, arrival, deadline, duration); } private void validateReservation(Plan plan, ReservationId resId, @@ -224,8 +235,12 @@ public void testSubmitReservationFailoverAndDelete() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); + // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -233,7 +248,6 @@ public void testSubmitReservationFailoverAndDelete() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); ReservationDefinition reservationDefinition = @@ -273,10 +287,14 @@ public void testFailoverAndSubmitReservation() throws Exception { explicitFailover(); addNodeCapacityToPlan(rm2, 102400, 100); + ClientRMService clientService = rm2.getClientRMService(); + + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); // create a reservation - ClientRMService clientService = rm2.getClientRMService(); - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -284,7 +302,6 @@ public void testFailoverAndSubmitReservation() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); ReservationDefinition reservationDefinition = @@ -304,8 +321,12 @@ public void testSubmitReservationFailoverAndUpdate() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); + // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -313,7 +334,6 @@ public void testSubmitReservationFailoverAndUpdate() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); ReservationDefinition reservationDefinition = @@ -353,8 +373,12 @@ public void testSubmitUpdateReservationFailoverAndDelete() throws Exception { ClientRMService clientService = rm1.getClientRMService(); + ReservationId reservationID = getNewReservation(clientService) + .getReservationId(); + // create a reservation - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationSubmissionRequest request = createReservationSubmissionRequest( + reservationID); ReservationSubmissionResponse response = null; try { response = clientService.submitReservation(request); @@ -362,7 +386,6 @@ public void testSubmitUpdateReservationFailoverAndDelete() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId reservationID = response.getReservationId(); Assert.assertNotNull(reservationID); LOG.info("Submit reservation response: " + reservationID); ReservationDefinition reservationDefinition = @@ -419,8 +442,12 @@ public void testReservationResizeAfterFailover() throws Exception { ClientRMService clientService = rm1.getClientRMService(); - // create 3 reservations - ReservationSubmissionRequest request = createReservationSubmissionRequest(); + ReservationId resID1 = getNewReservation(clientService) + .getReservationId(); + + // create a reservation + ReservationSubmissionRequest request = createReservationSubmissionRequest( + resID1); ReservationDefinition reservationDefinition = request.getReservationDefinition(); ReservationSubmissionResponse response = null; @@ -430,25 +457,30 @@ public void testReservationResizeAfterFailover() throws Exception { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId resID1 = response.getReservationId(); Assert.assertNotNull(resID1); LOG.info("Submit reservation response: " + resID1); + + ReservationId resID2 = getNewReservation(clientService) + .getReservationId(); + request.setReservationId(resID2); try { response = clientService.submitReservation(request); } catch (Exception e) { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId resID2 = response.getReservationId(); Assert.assertNotNull(resID2); LOG.info("Submit reservation response: " + resID2); + + ReservationId resID3 = getNewReservation(clientService) + .getReservationId(); + request.setReservationId(resID3); try { response = clientService.submitReservation(request); } catch (Exception e) { Assert.fail(e.getMessage()); } Assert.assertNotNull(response); - ReservationId resID3 = response.getReservationId(); Assert.assertNotNull(resID3); LOG.info("Submit reservation response: " + resID3); @@ -515,4 +547,18 @@ private void waitForReservationActivation(MockRM rm, } } + private GetNewReservationResponse getNewReservation(ClientRMService + clientRMService) { + GetNewReservationRequest newReservationRequest = GetNewReservationRequest + .newInstance(); + GetNewReservationResponse getNewReservationResponse = null; + try { + getNewReservationResponse = clientRMService.getNewReservation( + newReservationRequest); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + return getNewReservationResponse; + } + } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java index 4aef7ae..9ea1044 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java @@ -197,7 +197,8 @@ public static ReservationDefinition createSimpleReservationDefinition( } public static ReservationSubmissionRequest createSimpleReservationRequest( - int numContainers, long arrival, long deadline, long duration) { + ReservationId reservationId, int numContainers, long arrival, + long deadline, long duration) { // create a request with a single atomic ask ReservationRequest r = ReservationRequest.newInstance(Resource.newInstance(1024, 1), @@ -210,7 +211,7 @@ public static ReservationSubmissionRequest createSimpleReservationRequest( "testClientRMService#reservation"); ReservationSubmissionRequest request = ReservationSubmissionRequest.newInstance(rDef, - reservationQ); + reservationQ, reservationId); return request; } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesReservation.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesReservation.java index a60cf17..4bacccc 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesReservation.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesReservation.java @@ -19,8 +19,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.webapp; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import java.io.File; @@ -42,6 +41,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.security.authentication.server.AuthenticationFilter; import org.apache.hadoop.security.authentication.server.PseudoAuthenticationHandler; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewReservationRequest; import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.server.resourcemanager.MockNM; @@ -102,6 +102,8 @@ // This is what is used in the test resource files. private static final String DEFAULT_QUEUE = "dedicated"; private static final String LIST_RESERVATION_PATH = "reservation/list"; + private static final String GET_NEW_RESERVATION_PATH = + "reservation/new-reservation"; public static class GuiceServletConfig extends GuiceServletContextListener { @@ -330,12 +332,74 @@ public void tearDown() throws Exception { public void testSubmitReservation() throws Exception { rm.start(); setupCluster(100); - ReservationId rid = - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON); + + ReservationId rid = generateReservationId(1); + ClientResponse response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON, rid); + if (this.isAuthenticationEnabled()) { + assertTrue(isHttpSuccessResponse(response)); + verifyReservationCount(1); + } + rm.stop(); + } + + @Test + public void testSubmitDuplicateReservation() throws Exception { + rm.start(); + setupCluster(100); + + ReservationId rid = generateReservationId(1); + long currentTimestamp = clock.getTime() + MINIMUM_RESOURCE_DURATION; + ClientResponse response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON, currentTimestamp, "", + rid); + + // Make sure that the first submission is successful + if (this.isAuthenticationEnabled()) { + assertTrue(isHttpSuccessResponse(response)); + } + + response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON, currentTimestamp, "", + rid); + + // Make sure that the second submission is successful + if (this.isAuthenticationEnabled()) { + assertTrue(isHttpSuccessResponse(response)); + verifyReservationCount(1); + } + + rm.stop(); + } + + @Test + public void testSubmitDifferentReservationWithSameId() throws Exception { + rm.start(); + setupCluster(100); + + ReservationId rid = generateReservationId(1); + long currentTimestamp = clock.getTime() + MINIMUM_RESOURCE_DURATION; + ClientResponse response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON,currentTimestamp, + "res1", rid); + + // Make sure that the first submission is successful if (this.isAuthenticationEnabled()) { - assertNotNull(rid); + assertTrue(isHttpSuccessResponse(response)); } + + // Change the reservation definition. + response = reservationSubmissionTestHelper( + "reservation/submit", + MediaType.APPLICATION_JSON,currentTimestamp + MINIMUM_RESOURCE_DURATION, + "res1", rid); + + // Make sure that the second submission is unsuccessful + if (this.isAuthenticationEnabled()) { + assertTrue(!isHttpSuccessResponse(response)); + verifyReservationCount(1); + } + rm.stop(); } @@ -344,10 +408,13 @@ public void testFailedSubmitReservation() throws Exception { rm.start(); // setup a cluster too small to accept the reservation setupCluster(1); - ReservationId rid = - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON); - assertNull(rid); + + ReservationId rid = generateReservationId(1); + ClientResponse response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON, rid); + + assertTrue(!isHttpSuccessResponse(response)); + rm.stop(); } @@ -355,13 +422,14 @@ public void testFailedSubmitReservation() throws Exception { public void testUpdateReservation() throws JSONException, Exception { rm.start(); setupCluster(100); - ReservationId rid = - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON); + + ReservationId rid = generateReservationId(1); + ClientResponse response = reservationSubmissionTestHelper( + "reservation/submit", MediaType.APPLICATION_JSON, rid); if (this.isAuthenticationEnabled()) { - assertNotNull(rid); + assertTrue(isHttpSuccessResponse(response)); } - testUpdateReservationHelper("reservation/update", rid, + updateReservationTestHelper("reservation/update", rid, MediaType.APPLICATION_JSON); rm.stop(); @@ -373,11 +441,15 @@ public void testTimeIntervalRequestListReservation() throws Exception { setupCluster(100); long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("start-time", Long.toString((long) (time * 0.9))) @@ -410,11 +482,19 @@ public void testSameTimeIntervalRequestListReservation() throws Exception { long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + // If authentication is not enabled then id1 and id2 will be null + if (!this.isAuthenticationEnabled() && id1 == null && id2 == null) { + return; + } + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); String timeParam = Long.toString(time + MINIMUM_RESOURCE_DURATION / 2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) @@ -447,11 +527,14 @@ public void testInvalidTimeIntervalRequestListReservation() throws long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - ReservationId res1 = testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - ReservationId res2 = testSubmissionReservationHelper("reservation/submit", + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource; resource = constructWebResource(LIST_RESERVATION_PATH) @@ -483,11 +566,14 @@ public void testInvalidEndTimeRequestListReservation() throws Exception { long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("start-time", Long.toString((long) (time + @@ -520,11 +606,14 @@ public void testEmptyEndTimeRequestListReservation() throws Exception { long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("start-time", new Long((long) (time + @@ -556,11 +645,14 @@ public void testInvalidStartTimeRequestListReservation() throws Exception { long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("start-time", "-1") @@ -592,13 +684,16 @@ public void testEmptyStartTimeRequestListReservation() throws Exception { rm.start(); setupCluster(100); + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + long time = clock.getTime() + MINIMUM_RESOURCE_DURATION; - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, time, "res_1", 1); - testSubmissionReservationHelper("reservation/submit", + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, time, "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", MediaType.APPLICATION_JSON, time + MINIMUM_RESOURCE_DURATION, - "res_2", 2); + "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("end-time", new Long((long)(time + @@ -629,10 +724,13 @@ public void testQueueOnlyRequestListReservation() throws Exception { rm.start(); setupCluster(100); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_2", 2); + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("queue", DEFAULT_QUEUE); @@ -656,10 +754,13 @@ public void testEmptyQueueRequestListReservation() throws Exception { rm.start(); setupCluster(100); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_2", 2); + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH); @@ -673,10 +774,13 @@ public void testNonExistentQueueRequestListReservation() throws Exception { rm.start(); setupCluster(100); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_2", 2); + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("queue", DEFAULT_QUEUE + "_invalid"); @@ -691,10 +795,15 @@ public void testReservationIdRequestListReservation() throws Exception { rm.start(); setupCluster(100); - ReservationId id1 = testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_2", 2); + ReservationId id1 = generateReservationId(1); + ReservationId id2 = generateReservationId(2); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_2", id2); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("include-resource-allocations", "true") @@ -726,8 +835,10 @@ public void testInvalidReservationIdRequestListReservation() throws rm.start(); setupCluster(100); - ReservationId id1 = testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); + ReservationId id1 = generateReservationId(1); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("queue", DEFAULT_QUEUE); @@ -747,8 +858,9 @@ public void testIncludeResourceAllocations() throws Exception { rm.start(); setupCluster(100); - ReservationId id1 = testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); + ReservationId id1 = generateReservationId(1); + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("include-resource-allocations", "true") @@ -781,8 +893,10 @@ public void testExcludeResourceAllocations() throws Exception { rm.start(); setupCluster(100); - ReservationId id1 = testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON, clock.getTime(), "res_1", 1); + ReservationId id1 = generateReservationId(1); + + reservationSubmissionTestHelper("reservation/submit", + MediaType.APPLICATION_JSON, clock.getTime(), "res_1", id1); WebResource resource = constructWebResource(LIST_RESERVATION_PATH) .queryParam("include-resource-allocations", "false") @@ -818,40 +932,76 @@ public void testDeleteReservation() throws JSONException, Exception { rm.registerNode("127.0.0." + i + ":1234", 100 * 1024); amNodeManager.nodeHeartbeat(true); } - ReservationId rid = - testSubmissionReservationHelper("reservation/submit", - MediaType.APPLICATION_JSON); - if (this.isAuthenticationEnabled()) { - assertNotNull(rid); - } + + ReservationId rid = generateReservationId(1); + + reservationSubmissionTestHelper("reservation/submit", MediaType + .APPLICATION_JSON, rid); testDeleteReservationHelper("reservation/delete", rid, MediaType.APPLICATION_JSON); rm.stop(); } - private ReservationId testSubmissionReservationHelper(String path, - String media) throws Exception { + /** + * This method is used when the createReservation API isn't to be + * explicitly tested, but a reservation ID is still required. + * @param reservationId A long representing the reservation ID number to be + * generated. + * @return the object representing the reservation ID. + */ + private ReservationId generateReservationId(int reservationId) { + return ReservationId.newInstance(clock.getTime(), reservationId); + } + + private ReservationId getReservationIdTestHelper(String media) throws + Exception { + Thread.sleep(1000); + ClientResponse response = + constructWebResource(GET_NEW_RESERVATION_PATH).type(MediaType + .APPLICATION_JSON).accept(media).post(ClientResponse.class); + + if (!this.isAuthenticationEnabled()) { + assertEquals(Status.UNAUTHORIZED, response.getClientResponseStatus()); + return null; + } + + System.out.println("RESPONSE:" + response); + assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType()); + JSONObject json = response.getEntity(JSONObject.class); + + assertEquals("incorrect number of elements", 1, json.length()); + ReservationId rid = null; + try { + rid = ReservationId.parseReservationId(json.getString("reservation-id")); + } catch (JSONException j) { + // failure is possible and is checked outside + } + return rid; + } + + private ClientResponse reservationSubmissionTestHelper(String path, + String media, ReservationId reservationId) throws Exception { long arrival = clock.getTime() + MINIMUM_RESOURCE_DURATION; - return testSubmissionReservationHelper(path, media, arrival, "res_1", 1); + return reservationSubmissionTestHelper(path, media, arrival, "res_1", + reservationId); } - private ReservationId testSubmissionReservationHelper(String path, - String media, Long arrival, String reservationName, int expectedId) - throws Exception { + private ClientResponse reservationSubmissionTestHelper(String path, + String media, Long arrival, String reservationName, + ReservationId reservationId) throws Exception { String reservationJson = loadJsonFile("submit-reservation.json"); - String reservationJsonRequest = String.format(reservationJson, arrival, - arrival + MINIMUM_RESOURCE_DURATION, reservationName); + String reservationJsonRequest = String.format(reservationJson, + reservationId.toString(), arrival, arrival + MINIMUM_RESOURCE_DURATION, + reservationName); - return submitAndVerifyReservation(path, media, reservationJsonRequest, - expectedId); + return submitAndVerifyReservation(path, media, reservationJsonRequest); } - private ReservationId submitAndVerifyReservation(String path, String media, - String reservationJson, int expectedId) throws Exception { - + private ClientResponse submitAndVerifyReservation(String path, String media, + String reservationJson) throws Exception { JSONJAXBContext jc = new JSONJAXBContext(JSONConfiguration.mapped() .build(), ReservationSubmissionRequestInfo.class); @@ -867,25 +1017,12 @@ private ReservationId submitAndVerifyReservation(String path, String media, if (!this.isAuthenticationEnabled()) { assertEquals(Status.UNAUTHORIZED, response.getClientResponseStatus()); - return null; } - System.out.println("RESPONSE:" + response); - assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType()); - JSONObject json = response.getEntity(JSONObject.class); - - assertEquals("incorrect number of elements", 1, json.length()); - ReservationId rid = null; - try { - rid = ReservationId.parseReservationId(json.getString("reservation-id")); - assertEquals("incorrect return value", rid.getId(), expectedId); - } catch (JSONException j) { - // failure is possible and is checked outside - } - return rid; + return response; } - private void testUpdateReservationHelper(String path, + private void updateReservationTestHelper(String path, ReservationId reservationId, String media) throws JSONException, Exception { @@ -1002,6 +1139,25 @@ private JSONObject testListReservationHelper(WebResource resource, Status return response.getEntity(JSONObject.class); } + private void verifyReservationCount(int count) throws Exception { + WebResource resource = constructWebResource(LIST_RESERVATION_PATH) + .queryParam("queue", DEFAULT_QUEUE); + + JSONObject json = testListReservationHelper(resource); + + if (count == 1) { + // If there are any number other than one reservation, this will throw. + json.getJSONObject("reservations"); + } else { + JSONArray reservations = json.getJSONArray("reservations"); + assertTrue(reservations.length() == count); + } + } + + private boolean isHttpSuccessResponse(ClientResponse response) { + return (response.getStatus() / 100) == 2; + } + private void setupCluster(int nodes) throws Exception { for (int i = 0; i < nodes; i++) { MockNM amNodeManager = diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/resources/submit-reservation.json hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/resources/submit-reservation.json index 0a243a2..580c599 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/resources/submit-reservation.json +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/resources/submit-reservation.json @@ -1,5 +1,6 @@ { "queue" : "dedicated", + "reservation-id" : "%s", "reservation-definition" : { "arrival" : %s, "deadline" : %s, diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ReservationSystem.md hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ReservationSystem.md index eda8d4d..51d2b97 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ReservationSystem.md +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ReservationSystem.md @@ -39,7 +39,9 @@ Flow of a Reservation With reference to the figure above, a typical reservation proceeds as follows: - * **Step 1** The user (or an automated tool on its behalf) submit a reservation request specified by the Reservation Definition Language (RDL). This describes the user need for resources over-time (e.g., a skyline of resources) and temporal constraints (e.g., deadline). This can be done both programmatically through the usual Client-to-RM protocols or via the REST api of the RM. + * **Step 0** The user (or an automated tool on its behalf) submits a reservation creation request, and receives a response containing the ReservationId. + + * **Step 1** The user (or an automated tool on its behalf) submits a reservation request specified by the Reservation Definition Language (RDL) and ReservationId retrieved from the previous step. This describes the user need for resources over-time (e.g., a skyline of resources) and temporal constraints (e.g., deadline). This can be done both programmatically through the usual Client-to-RM protocols or via the REST api of the RM. If a reservation is submitted with the same ReservationId, and the RDL is the same, a new reservation will not be created and the request will be successful. If the RDL is different, the reservation will be rejected, and the request will be unsuccessful. * **Step 2** The ReservationSystem leverages a ReservationAgent (GREE in the figure) to find a plausible allocation for the reservation in the Plan, a data structure tracking all reservation currently accepted and the available resources in the system. diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ResourceManagerRest.md hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ResourceManagerRest.md index dd6ac04..58c54e5 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ResourceManagerRest.md +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ResourceManagerRest.md @@ -34,6 +34,7 @@ ResourceManager REST API's. * [Cluster Application Priority API](#Cluster_Application_Priority_API) * [Cluster Delegation Tokens API](#Cluster_Delegation_Tokens_API) * [Cluster Reservation API List](#Cluster_Reservation_API_List) +* [Cluster Reservation API Create](#Cluster_Reservation_API_Create) * [Cluster Reservation API Submit](#Cluster_Reservation_API_Submit) * [Cluster Reservation API Update](#Cluster_Reservation_API_Update) * [Cluster Reservation API Delete](#Cluster_Reservation_API_Delete) @@ -3385,10 +3386,82 @@ Response Body: ``` +Cluster Reservation API Create +--------------------------- + +With the New Reservation API, you can obtain a reservation-id which can then be used as part of the [Cluster Reservation API Submit](#Cluster_Reservation_API_Submit) to submit reservations. + +This feature is currently in the alpha stage and may change in the future. + +### URI + + * http://