goiri commented on a change in pull request #3135:
URL: https://github.com/apache/hadoop/pull/3135#discussion_r658485767
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterYarnClientUtils.java
##########
@@ -52,4 +63,126 @@ public static GetClusterMetricsResponse merge(
}
return GetClusterMetricsResponse.newInstance(tmp);
}
+
+ /**
+ * Merges a list of ApplicationReports grouping by ApplicationId.
+ * Our current policy is to merge the application reports from the reachable
+ * SubClusters.
+ * @param responses a list of ApplicationResponse to merge
+ * @param returnPartialResult if the merge ApplicationReports should contain
+ * partial result or not
+ * @return the merged ApplicationsResponse
+ */
+ public static GetApplicationsResponse mergeApplications(
+ Collection<GetApplicationsResponse> responses,
+ boolean returnPartialResult){
+ Map<ApplicationId, ApplicationReport> federationAM = new HashMap<>();
+ Map<ApplicationId, ApplicationReport> federationUAMSum = new HashMap<>();
+
+ for (GetApplicationsResponse appResponse : responses){
+ for (ApplicationReport appReport : appResponse.getApplicationList()){
+ ApplicationId appId = appReport.getApplicationId();
+ // Check if this ApplicationReport is an AM
+ if (appReport.getHost() != null) {
+ // Insert in the list of AM
+ federationAM.put(appId, appReport);
+ // Check if there are any UAM found before
+ if (federationUAMSum.containsKey(appId)) {
+ // Merge the current AM with the found UAM
+ mergeAMWithUAM(appReport, federationUAMSum.get(appId));
+ // Remove the sum of the UAMs
+ federationUAMSum.remove(appId);
+ }
+ // This ApplicationReport is an UAM
+ } else if (federationAM.containsKey(appId)) {
+ // Merge the current UAM with its own AM
+ mergeAMWithUAM(federationAM.get(appId), appReport);
+ } else if (federationUAMSum.containsKey(appId)) {
+ // Merge the current UAM with its own UAM and update the list of UAM
+ federationUAMSum.put(appId,
+ mergeUAMWithUAM(federationUAMSum.get(appId), appReport));
Review comment:
Extract the result of the merge.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterYarnClientUtils.java
##########
@@ -52,4 +63,126 @@ public static GetClusterMetricsResponse merge(
}
return GetClusterMetricsResponse.newInstance(tmp);
}
+
+ /**
+ * Merges a list of ApplicationReports grouping by ApplicationId.
+ * Our current policy is to merge the application reports from the reachable
+ * SubClusters.
+ * @param responses a list of ApplicationResponse to merge
+ * @param returnPartialResult if the merge ApplicationReports should contain
+ * partial result or not
+ * @return the merged ApplicationsResponse
+ */
+ public static GetApplicationsResponse mergeApplications(
+ Collection<GetApplicationsResponse> responses,
+ boolean returnPartialResult){
+ Map<ApplicationId, ApplicationReport> federationAM = new HashMap<>();
+ Map<ApplicationId, ApplicationReport> federationUAMSum = new HashMap<>();
+
+ for (GetApplicationsResponse appResponse : responses){
+ for (ApplicationReport appReport : appResponse.getApplicationList()){
+ ApplicationId appId = appReport.getApplicationId();
+ // Check if this ApplicationReport is an AM
+ if (appReport.getHost() != null) {
+ // Insert in the list of AM
+ federationAM.put(appId, appReport);
+ // Check if there are any UAM found before
+ if (federationUAMSum.containsKey(appId)) {
+ // Merge the current AM with the found UAM
+ mergeAMWithUAM(appReport, federationUAMSum.get(appId));
+ // Remove the sum of the UAMs
+ federationUAMSum.remove(appId);
+ }
+ // This ApplicationReport is an UAM
+ } else if (federationAM.containsKey(appId)) {
+ // Merge the current UAM with its own AM
+ mergeAMWithUAM(federationAM.get(appId), appReport);
+ } else if (federationUAMSum.containsKey(appId)) {
+ // Merge the current UAM with its own UAM and update the list of UAM
+ federationUAMSum.put(appId,
+ mergeUAMWithUAM(federationUAMSum.get(appId), appReport));
+ } else {
+ // Insert in the list of UAM
+ federationUAMSum.put(appId, appReport);
+ }
+ }
+ }
+ // Check the remaining UAMs are depending or not from federation
+ for (ApplicationReport appReport : federationUAMSum.values()) {
+ if (mergeUamToReport(appReport.getName(), returnPartialResult)) {
+ federationAM.put(appReport.getApplicationId(), appReport);
+ }
+ }
+
+ return GetApplicationsResponse.newInstance(federationAM.values());
+ }
+
+ private static ApplicationReport mergeUAMWithUAM(ApplicationReport uam1,
+ ApplicationReport uam2){
+ uam1.setName(PARTIAL_REPORT + uam1.getApplicationId());
+ mergeAMWithUAM(uam1, uam1);
+ mergeAMWithUAM(uam1, uam2);
+ return uam1;
+ }
+
+ private static void mergeAMWithUAM(ApplicationReport am,
+ ApplicationReport uam){
+ ApplicationResourceUsageReport amResourceReport =
+ am.getApplicationResourceUsageReport();
+
+ ApplicationResourceUsageReport uamResourceReport =
+ uam.getApplicationResourceUsageReport();
+
+ amResourceReport.setNumUsedContainers(
+ amResourceReport.getNumUsedContainers() +
+ uamResourceReport.getNumUsedContainers());
+
+ amResourceReport.setNumReservedContainers(
+ amResourceReport.getNumReservedContainers() +
+ uamResourceReport.getNumReservedContainers());
+
+ amResourceReport.setUsedResources(Resources.add(
+ amResourceReport.getUsedResources(),
+ uamResourceReport.getUsedResources()));
+
+ amResourceReport.setReservedResources(Resources.add(
+ amResourceReport.getReservedResources(),
+ uamResourceReport.getReservedResources()));
+
+ amResourceReport.setNeededResources(Resources.add(
+ amResourceReport.getNeededResources(),
+ uamResourceReport.getNeededResources()));
+
+ amResourceReport.setMemorySeconds(
+ amResourceReport.getMemorySeconds() +
+ uamResourceReport.getMemorySeconds());
+
+ amResourceReport.setVcoreSeconds(
+ amResourceReport.getVcoreSeconds() +
+ uamResourceReport.getVcoreSeconds());
+
+ amResourceReport.setQueueUsagePercentage(
+ amResourceReport.getQueueUsagePercentage() +
+ uamResourceReport.getQueueUsagePercentage());
+
+ amResourceReport.setClusterUsagePercentage(
+ amResourceReport.getClusterUsagePercentage() +
+ uamResourceReport.getClusterUsagePercentage());
+
+ am.getApplicationTags().addAll(uam.getApplicationTags());
+ }
+
+ /**
+ * Returns whether or not to add an unmanaged application to the report.
+ * @param appName Application Name
+ * @param returnPartialResult if the merge ApplicationReports should contain
+ * partial result or not
+ */
+ private static boolean mergeUamToReport(String appName,
+ boolean returnPartialResult){
+
+ return returnPartialResult || (appName != null &&
Review comment:
To make it more readable you could do somethin like:
```
if (returnPartialResult) {
return true;
}
if (appName == null) {
return false;
}
```
...
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/conf/TestYarnConfigurationFields.java
##########
@@ -184,6 +184,8 @@ public void initializeMemberVariables() {
configurationPrefixToSkipCompare
.add(YarnConfiguration.ROUTER_CLIENTRM_SUBMIT_RETRY);
+ configurationPrefixToSkipCompare
+ .add(YarnConfiguration.ROUTER_CLIENTRM_PARTIAL_RESULTS_ENABLED);
Review comment:
Match the indentation with the previous one.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestRouterYarnClientUtils.java
##########
@@ -54,4 +64,69 @@ public GetClusterMetricsResponse
getClusterMetricsResponse(int value) {
metrics.setNumNodeManagers(value);
return GetClusterMetricsResponse.newInstance(metrics);
}
+
+ /**
+ * This test validates the correctness of
+ * RouterYarnClientUtils#mergeApplications.
+ */
+ @Test
+ public void testMergeApplications() {
+ ArrayList<GetApplicationsResponse> responses = new ArrayList<>();
+ responses.add(getApplicationsResponse(1));
+ responses.add(getApplicationsResponse(2));
+ GetApplicationsResponse result = RouterYarnClientUtils.
+ mergeApplications(responses, false);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(2, result.getApplicationList().size());
+ }
+
+ /**
+ * This generates a GetApplicationsResponse with 2 applications with
+ * same ApplicationId. One of them is added with host value equal to
+ * null to validate unmanaged application merge with managed application.
+ * @param value Used as Id in ApplicationId
+ * @return GetApplicationsResponse
+ */
+ private GetApplicationsResponse getApplicationsResponse(int value) {
+ List<ApplicationReport> applications = new ArrayList<>();
+
+ //Add managed application to list
+ ApplicationId appId = ApplicationId.newInstance(1234,
+ value);
+ Resource resource = Resource.newInstance(1024, 1);
+ ApplicationResourceUsageReport appResourceUsageReport =
+ ApplicationResourceUsageReport.newInstance(
+ 1, 2, resource, resource,
+ resource, null, 0.1f,
+ 0.1f, null);
+
+ ApplicationReport appReport = ApplicationReport.newInstance(
+ appId, ApplicationAttemptId.newInstance(appId,
+ 1),
Review comment:
Single line
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
##########
@@ -531,4 +536,110 @@ public void testGetClusterMetricsRequest() throws
YarnException, IOException {
GetClusterMetricsResponse.class);
Assert.assertEquals(true, clusterMetrics.isEmpty());
}
+
+ /**
+ * This test validates the correctness of
+ * GetApplicationsResponse in case the
+ * application exists in the cluster.
+ */
+ @Test
+ public void testGetApplicationsResponse()
+ throws YarnException, IOException, InterruptedException {
+ LOG.info("Test FederationClientInterceptor: " +
+ "Get Applications Response");
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("MockApp");
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case of
+ * empty request.
+ */
+ @Test
+ public void testGetApplicationsNullRequest() throws Exception {
+ LOG.info("Test FederationClientInterceptor : Get Applications request");
+ LambdaTestUtils.intercept(YarnException.class,
+ "Missing getApplications request.",
+ () -> interceptor.getApplications(null));
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case applications
+ * with given type does not exist.
+ */
+ @Test
+ public void testGetApplicationsApplicationTypeNotExists() throws Exception{
+ LOG.info("Test FederationClientInterceptor :" +
+ " Application with type does not exist");
+
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("SPARK");
+
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ Assert.assertTrue(responseGet.getApplicationList().isEmpty());
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case applications
+ * with given YarnApplicationState does not exist.
+ */
+ @Test
+ public void testGetApplicationsApplicationStateNotExists() throws Exception{
+ LOG.info("Test FederationClientInterceptor :" +
+ " Application with state does not exist");
+
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
Review comment:
The indentation should be 4 spaces after break line.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestRouterYarnClientUtils.java
##########
@@ -54,4 +64,69 @@ public GetClusterMetricsResponse
getClusterMetricsResponse(int value) {
metrics.setNumNodeManagers(value);
return GetClusterMetricsResponse.newInstance(metrics);
}
+
+ /**
+ * This test validates the correctness of
+ * RouterYarnClientUtils#mergeApplications.
+ */
+ @Test
+ public void testMergeApplications() {
+ ArrayList<GetApplicationsResponse> responses = new ArrayList<>();
+ responses.add(getApplicationsResponse(1));
+ responses.add(getApplicationsResponse(2));
+ GetApplicationsResponse result = RouterYarnClientUtils.
+ mergeApplications(responses, false);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(2, result.getApplicationList().size());
+ }
+
+ /**
+ * This generates a GetApplicationsResponse with 2 applications with
+ * same ApplicationId. One of them is added with host value equal to
+ * null to validate unmanaged application merge with managed application.
+ * @param value Used as Id in ApplicationId
+ * @return GetApplicationsResponse
+ */
+ private GetApplicationsResponse getApplicationsResponse(int value) {
+ List<ApplicationReport> applications = new ArrayList<>();
+
+ //Add managed application to list
+ ApplicationId appId = ApplicationId.newInstance(1234,
+ value);
+ Resource resource = Resource.newInstance(1024, 1);
+ ApplicationResourceUsageReport appResourceUsageReport =
+ ApplicationResourceUsageReport.newInstance(
+ 1, 2, resource, resource,
+ resource, null, 0.1f,
+ 0.1f, null);
+
+ ApplicationReport appReport = ApplicationReport.newInstance(
+ appId, ApplicationAttemptId.newInstance(appId,
+ 1),
+ "user", "queue", "appname", "host",
+ 124, null, YarnApplicationState.RUNNING,
+ "diagnostics", "url", 0, 0,
+ 0, FinalApplicationStatus.SUCCEEDED,
+ appResourceUsageReport,
+ "N/A", 0.53789f, "YARN",
+ null);
+
+ //Add unmanaged application to list
+ ApplicationId appId2 = ApplicationId.newInstance(1234,
+ value);
+ ApplicationReport appReport2 = ApplicationReport.newInstance(
+ appId2, ApplicationAttemptId.newInstance(appId2,
Review comment:
newInstace() in a single line.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterYarnClientUtils.java
##########
@@ -52,4 +63,126 @@ public static GetClusterMetricsResponse merge(
}
return GetClusterMetricsResponse.newInstance(tmp);
}
+
+ /**
+ * Merges a list of ApplicationReports grouping by ApplicationId.
+ * Our current policy is to merge the application reports from the reachable
+ * SubClusters.
+ * @param responses a list of ApplicationResponse to merge
+ * @param returnPartialResult if the merge ApplicationReports should contain
+ * partial result or not
+ * @return the merged ApplicationsResponse
+ */
+ public static GetApplicationsResponse mergeApplications(
Review comment:
This looks like visible just for testing, add the @VisibleForTesting
annotation.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterYarnClientUtils.java
##########
@@ -52,4 +63,126 @@ public static GetClusterMetricsResponse merge(
}
return GetClusterMetricsResponse.newInstance(tmp);
}
+
+ /**
+ * Merges a list of ApplicationReports grouping by ApplicationId.
+ * Our current policy is to merge the application reports from the reachable
+ * SubClusters.
+ * @param responses a list of ApplicationResponse to merge
+ * @param returnPartialResult if the merge ApplicationReports should contain
+ * partial result or not
+ * @return the merged ApplicationsResponse
+ */
+ public static GetApplicationsResponse mergeApplications(
+ Collection<GetApplicationsResponse> responses,
+ boolean returnPartialResult){
+ Map<ApplicationId, ApplicationReport> federationAM = new HashMap<>();
+ Map<ApplicationId, ApplicationReport> federationUAMSum = new HashMap<>();
+
+ for (GetApplicationsResponse appResponse : responses){
+ for (ApplicationReport appReport : appResponse.getApplicationList()){
+ ApplicationId appId = appReport.getApplicationId();
+ // Check if this ApplicationReport is an AM
+ if (appReport.getHost() != null) {
+ // Insert in the list of AM
+ federationAM.put(appId, appReport);
+ // Check if there are any UAM found before
+ if (federationUAMSum.containsKey(appId)) {
+ // Merge the current AM with the found UAM
+ mergeAMWithUAM(appReport, federationUAMSum.get(appId));
+ // Remove the sum of the UAMs
+ federationUAMSum.remove(appId);
+ }
+ // This ApplicationReport is an UAM
+ } else if (federationAM.containsKey(appId)) {
+ // Merge the current UAM with its own AM
+ mergeAMWithUAM(federationAM.get(appId), appReport);
+ } else if (federationUAMSum.containsKey(appId)) {
+ // Merge the current UAM with its own UAM and update the list of UAM
+ federationUAMSum.put(appId,
+ mergeUAMWithUAM(federationUAMSum.get(appId), appReport));
+ } else {
+ // Insert in the list of UAM
+ federationUAMSum.put(appId, appReport);
+ }
+ }
+ }
+ // Check the remaining UAMs are depending or not from federation
+ for (ApplicationReport appReport : federationUAMSum.values()) {
+ if (mergeUamToReport(appReport.getName(), returnPartialResult)) {
+ federationAM.put(appReport.getApplicationId(), appReport);
+ }
+ }
+
+ return GetApplicationsResponse.newInstance(federationAM.values());
+ }
+
+ private static ApplicationReport mergeUAMWithUAM(ApplicationReport uam1,
+ ApplicationReport uam2){
+ uam1.setName(PARTIAL_REPORT + uam1.getApplicationId());
Review comment:
Can we test for this kind of name?
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java
##########
@@ -599,10 +604,45 @@ public GetApplicationReportResponse getApplicationReport(
return response;
}
+ /**
+ * The Yarn Router will forward the request to all the Yarn RMs in parallel,
+ * after that it will group all the ApplicationReports by the ApplicationId.
+ *
+ * Possible failure:
+ *
+ * Client: identical behavior as {@code ClientRMService}.
+ *
+ * Router: the Client will timeout and resubmit the request.
+ *
+ * ResourceManager: the Router calls each Yarn RM in parallel. In case a
+ * Yarn RM fails, a single call will timeout. However the Router will
+ * merge the ApplicationReports it got, and provides a partial list to
+ * the client.
+ *
+ * State Store: the Router will timeout and it will retry depending on the
+ * FederationFacade settings - if the failure happened before the select
+ * operation.
+ */
@Override
public GetApplicationsResponse getApplications(GetApplicationsRequest
request)
throws YarnException, IOException {
- throw new NotImplementedException("Code is not implemented");
+ if (request == null) {
+ RouterServerUtil.logAndThrowException(
+ "Missing getApplications request.",
+ null);
+ }
+ Map<SubClusterId, SubClusterInfo> subclusters =
+ federationFacade.getSubClusters(true);
+ ClientMethod remoteMethod = new ClientMethod("getApplications",
+ new Class[] {GetApplicationsRequest.class}, new Object[] {request});
+ ArrayList<SubClusterId> clusterIds = new ArrayList<>(subclusters.keySet());
+ Map<SubClusterId, GetApplicationsResponse> applications =
+ invokeConcurrent(clusterIds, remoteMethod,
Review comment:
This probably fits in a single line
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
##########
@@ -531,4 +536,110 @@ public void testGetClusterMetricsRequest() throws
YarnException, IOException {
GetClusterMetricsResponse.class);
Assert.assertEquals(true, clusterMetrics.isEmpty());
}
+
+ /**
+ * This test validates the correctness of
+ * GetApplicationsResponse in case the
+ * application exists in the cluster.
+ */
+ @Test
+ public void testGetApplicationsResponse()
+ throws YarnException, IOException, InterruptedException {
+ LOG.info("Test FederationClientInterceptor: " +
+ "Get Applications Response");
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("MockApp");
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case of
+ * empty request.
+ */
+ @Test
+ public void testGetApplicationsNullRequest() throws Exception {
+ LOG.info("Test FederationClientInterceptor : Get Applications request");
+ LambdaTestUtils.intercept(YarnException.class,
+ "Missing getApplications request.",
+ () -> interceptor.getApplications(null));
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case applications
+ * with given type does not exist.
+ */
+ @Test
+ public void testGetApplicationsApplicationTypeNotExists() throws Exception{
+ LOG.info("Test FederationClientInterceptor :" +
Review comment:
```suggestion
LOG.info("Test FederationClientInterceptor: Application with type does
not exist");
```
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestRouterYarnClientUtils.java
##########
@@ -54,4 +64,69 @@ public GetClusterMetricsResponse
getClusterMetricsResponse(int value) {
metrics.setNumNodeManagers(value);
return GetClusterMetricsResponse.newInstance(metrics);
}
+
+ /**
+ * This test validates the correctness of
+ * RouterYarnClientUtils#mergeApplications.
+ */
+ @Test
+ public void testMergeApplications() {
+ ArrayList<GetApplicationsResponse> responses = new ArrayList<>();
+ responses.add(getApplicationsResponse(1));
+ responses.add(getApplicationsResponse(2));
+ GetApplicationsResponse result = RouterYarnClientUtils.
+ mergeApplications(responses, false);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(2, result.getApplicationList().size());
+ }
+
+ /**
+ * This generates a GetApplicationsResponse with 2 applications with
+ * same ApplicationId. One of them is added with host value equal to
+ * null to validate unmanaged application merge with managed application.
+ * @param value Used as Id in ApplicationId
+ * @return GetApplicationsResponse
+ */
+ private GetApplicationsResponse getApplicationsResponse(int value) {
+ List<ApplicationReport> applications = new ArrayList<>();
+
+ //Add managed application to list
+ ApplicationId appId = ApplicationId.newInstance(1234,
+ value);
+ Resource resource = Resource.newInstance(1024, 1);
+ ApplicationResourceUsageReport appResourceUsageReport =
+ ApplicationResourceUsageReport.newInstance(
+ 1, 2, resource, resource,
+ resource, null, 0.1f,
+ 0.1f, null);
+
+ ApplicationReport appReport = ApplicationReport.newInstance(
+ appId, ApplicationAttemptId.newInstance(appId,
+ 1),
+ "user", "queue", "appname", "host",
+ 124, null, YarnApplicationState.RUNNING,
+ "diagnostics", "url", 0, 0,
+ 0, FinalApplicationStatus.SUCCEEDED,
+ appResourceUsageReport,
+ "N/A", 0.53789f, "YARN",
+ null);
+
+ //Add unmanaged application to list
+ ApplicationId appId2 = ApplicationId.newInstance(1234,
Review comment:
Single line
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
##########
@@ -531,4 +536,110 @@ public void testGetClusterMetricsRequest() throws
YarnException, IOException {
GetClusterMetricsResponse.class);
Assert.assertEquals(true, clusterMetrics.isEmpty());
}
+
+ /**
+ * This test validates the correctness of
+ * GetApplicationsResponse in case the
+ * application exists in the cluster.
+ */
+ @Test
+ public void testGetApplicationsResponse()
+ throws YarnException, IOException, InterruptedException {
+ LOG.info("Test FederationClientInterceptor: " +
+ "Get Applications Response");
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("MockApp");
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case of
+ * empty request.
+ */
+ @Test
+ public void testGetApplicationsNullRequest() throws Exception {
+ LOG.info("Test FederationClientInterceptor : Get Applications request");
+ LambdaTestUtils.intercept(YarnException.class,
+ "Missing getApplications request.",
+ () -> interceptor.getApplications(null));
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case applications
+ * with given type does not exist.
+ */
+ @Test
+ public void testGetApplicationsApplicationTypeNotExists() throws Exception{
+ LOG.info("Test FederationClientInterceptor :" +
+ " Application with type does not exist");
+
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("SPARK");
+
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ Assert.assertTrue(responseGet.getApplicationList().isEmpty());
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case applications
+ * with given YarnApplicationState does not exist.
+ */
+ @Test
+ public void testGetApplicationsApplicationStateNotExists() throws Exception{
+ LOG.info("Test FederationClientInterceptor :" +
+ " Application with state does not exist");
+
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ EnumSet<YarnApplicationState> applicationStates = EnumSet.noneOf(
+ YarnApplicationState.class);
+ applicationStates.add(YarnApplicationState.KILLED);
+
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(applicationStates);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ Assert.assertEquals(0, responseGet.getApplicationList().size());
Review comment:
isEmpty()
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java
##########
@@ -599,10 +604,45 @@ public GetApplicationReportResponse getApplicationReport(
return response;
}
+ /**
+ * The Yarn Router will forward the request to all the Yarn RMs in parallel,
+ * after that it will group all the ApplicationReports by the ApplicationId.
+ *
+ * Possible failure:
+ *
+ * Client: identical behavior as {@code ClientRMService}.
+ *
+ * Router: the Client will timeout and resubmit the request.
+ *
+ * ResourceManager: the Router calls each Yarn RM in parallel. In case a
+ * Yarn RM fails, a single call will timeout. However the Router will
+ * merge the ApplicationReports it got, and provides a partial list to
+ * the client.
+ *
+ * State Store: the Router will timeout and it will retry depending on the
+ * FederationFacade settings - if the failure happened before the select
+ * operation.
+ */
@Override
public GetApplicationsResponse getApplications(GetApplicationsRequest
request)
throws YarnException, IOException {
- throw new NotImplementedException("Code is not implemented");
+ if (request == null) {
+ RouterServerUtil.logAndThrowException(
+ "Missing getApplications request.",
+ null);
+ }
+ Map<SubClusterId, SubClusterInfo> subclusters =
+ federationFacade.getSubClusters(true);
+ ClientMethod remoteMethod = new ClientMethod("getApplications",
+ new Class[] {GetApplicationsRequest.class}, new Object[] {request});
+ ArrayList<SubClusterId> clusterIds = new ArrayList<>(subclusters.keySet());
Review comment:
I'm pretty sure we are doing this conversion to list a bunch of times,
having one that takes Collection would remove some noise.
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
##########
@@ -531,4 +536,110 @@ public void testGetClusterMetricsRequest() throws
YarnException, IOException {
GetClusterMetricsResponse.class);
Assert.assertEquals(true, clusterMetrics.isEmpty());
}
+
+ /**
+ * This test validates the correctness of
+ * GetApplicationsResponse in case the
+ * application exists in the cluster.
+ */
+ @Test
+ public void testGetApplicationsResponse()
+ throws YarnException, IOException, InterruptedException {
+ LOG.info("Test FederationClientInterceptor: " +
+ "Get Applications Response");
+ ApplicationId appId =
+ ApplicationId.newInstance(System.currentTimeMillis(), 1);
+
+ SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
+ SubmitApplicationResponse response =
interceptor.submitApplication(request);
+
+ Assert.assertNotNull(response);
+ Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));
+
+ Set<String> appTypes = Collections.singleton("MockApp");
+ GetApplicationsRequest requestGet =
+ GetApplicationsRequest.newInstance(appTypes);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(requestGet);
+
+ Assert.assertNotNull(responseGet);
+ }
+
+ /**
+ * This test validates
+ * the correctness of GetApplicationsResponse in case of
+ * empty request.
+ */
+ @Test
+ public void testGetApplicationsNullRequest() throws Exception {
+ LOG.info("Test FederationClientInterceptor : Get Applications request");
Review comment:
```suggestion
LOG.info("Test FederationClientInterceptor: Get Applications request");
```
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
##########
@@ -531,4 +536,107 @@ public void testGetClusterMetricsRequest() throws
YarnException, IOException {
GetClusterMetricsResponse.class);
Assert.assertEquals(true, clusterMetrics.isEmpty());
}
+
+ /**
+ * This test validates the correctness of
+ * GetApplicationsResponse in case the
+ * application exists in the cluster.
+ */
+ @Test
+ public void testGetApplicationsResponse()
+ throws YarnException, IOException, InterruptedException {
+ LOG.info("Test FederationClientInterceptor: " +
Review comment:
I mean that putting the whole thing would fit in 80:
```
LOG.info("Test FederationClientInterceptor: Get Applications Response");
```
##########
File path:
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestRouterYarnClientUtils.java
##########
@@ -54,4 +64,69 @@ public GetClusterMetricsResponse
getClusterMetricsResponse(int value) {
metrics.setNumNodeManagers(value);
return GetClusterMetricsResponse.newInstance(metrics);
}
+
+ /**
+ * This test validates the correctness of
+ * RouterYarnClientUtils#mergeApplications.
+ */
+ @Test
+ public void testMergeApplications() {
+ ArrayList<GetApplicationsResponse> responses = new ArrayList<>();
+ responses.add(getApplicationsResponse(1));
+ responses.add(getApplicationsResponse(2));
+ GetApplicationsResponse result = RouterYarnClientUtils.
+ mergeApplications(responses, false);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(2, result.getApplicationList().size());
+ }
+
+ /**
+ * This generates a GetApplicationsResponse with 2 applications with
+ * same ApplicationId. One of them is added with host value equal to
+ * null to validate unmanaged application merge with managed application.
+ * @param value Used as Id in ApplicationId
+ * @return GetApplicationsResponse
+ */
+ private GetApplicationsResponse getApplicationsResponse(int value) {
+ List<ApplicationReport> applications = new ArrayList<>();
+
+ //Add managed application to list
+ ApplicationId appId = ApplicationId.newInstance(1234,
Review comment:
Single line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-issues-h...@hadoop.apache.org
