goiri commented on a change in pull request #3135:
URL: https://github.com/apache/hadoop/pull/3135#discussion_r658080890
##########
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 +65,110 @@ 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)) {
Review comment:
```suggestion
} else if (federationAM.containsKey(appId)) {
```
And then the rest of the elses would go with this.
##########
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: " +
+ "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");
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(
+ GetApplicationsRequest.
+ newInstance(appTypes));
+
+ Assert.assertNotNull(responseGet);
+ Assert.assertEquals(0, responseGet.getApplicationList().size());
+ }
+
+ /**
+ * 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);
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(
+ GetApplicationsRequest.
Review comment:
Extract
##########
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 +65,110 @@ 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 (returnPartialResult || appReport.getName() != null
Review comment:
This if is a little convoluted.
Probably making it a static method would help readability.
##########
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 +607,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,
+ GetApplicationsResponse.class);
+
+ //Merge the Application Reports
Review comment:
```suggestion
// Merge the Application Reports
```
##########
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:
Does this fit in 80 chars?
##########
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: " +
+ "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");
+
+ GetApplicationsResponse responseGet =
+ interceptor.getApplications(
+ GetApplicationsRequest.
+ newInstance(appTypes));
+
+ Assert.assertNotNull(responseGet);
+ Assert.assertEquals(0, responseGet.getApplicationList().size());
Review comment:
```suggestion
Assert.assertTrue(responseGet.getApplicationList().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 +607,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.",
Review comment:
I don't think the spacing follows the right convention.
The checkstyle should flag all this.
##########
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 +65,110 @@ 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 (returnPartialResult || appReport.getName() != null
+ && !(appReport.getName()
+ .startsWith(UnmanagedApplicationManager.APP_NAME)
+ || appReport.getName()
+ .startsWith(PARTIAL_REPORT))) {
+ federationAM.put(appReport.getApplicationId(), appReport);
+ }
+ }
+
+ List<ApplicationReport> appList = new ArrayList<>(federationAM.values());
+ return GetApplicationsResponse.newInstance(appList);
+ }
+
+ 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 resourceUsageReport =
+ am.getApplicationResourceUsageReport();
+ resourceUsageReport.setNumUsedContainers(
+ resourceUsageReport.getNumUsedContainers() +
+ uam.getApplicationResourceUsageReport()
Review comment:
The readability is not very good here.
Maybe extracting the report for uam would help.
##########
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: " +
+ "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));
Review comment:
The indentation looks crooked.
Check the checkstyle report from Yetus.
##########
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 +65,110 @@ 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 (returnPartialResult || appReport.getName() != null
+ && !(appReport.getName()
+ .startsWith(UnmanagedApplicationManager.APP_NAME)
+ || appReport.getName()
+ .startsWith(PARTIAL_REPORT))) {
+ federationAM.put(appReport.getApplicationId(), appReport);
+ }
+ }
+
+ List<ApplicationReport> appList = new ArrayList<>(federationAM.values());
+ return GetApplicationsResponse.newInstance(appList);
Review comment:
Probably you could extend GetApplicationsResponse.newInstance() to take
a Collection<ApplicationReport> and do the new ArrayList there.
##########
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,62 @@ 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());
+ }
+
+ private GetApplicationsResponse getApplicationsResponse(int value) {
Review comment:
javadoc would help. Specially defining value.
##########
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
##########
@@ -196,6 +197,13 @@ public void init(String userName) {
clientRMProxies =
new ConcurrentHashMap<SubClusterId, ApplicationClientProtocol>();
routerMetrics = RouterMetrics.getMetrics();
+
+ returnPartialReport =
Review comment:
The lines could be split in a more friendly way:
```suggestion
returnPartialReport = conf.getBoolean(
YarnConfiguration.ROUTER_CLIENTRM_PARTIAL_RESULTS_ENABLED,
YarnConfiguration.DEFAULT_ROUTER_CLIENTRM_PARTIAL_RESULTS_ENABLED);
```
##########
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,62 @@ 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());
+ }
+
+ private GetApplicationsResponse getApplicationsResponse(int value) {
+ List<ApplicationReport> applications = new ArrayList<>();
+
+ //Add managed application to list
+ ApplicationId applicationId = 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 newApplicationReport = ApplicationReport.newInstance(
+ applicationId, ApplicationAttemptId.newInstance(applicationId,
+ 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 applicationId2 = ApplicationId.newInstance(1234,
Review comment:
You could use shorter names like appId2
##########
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 +65,110 @@ 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){
Review comment:
```suggestion
for (GetApplicationsResponse appResponse : responses){
```
--
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
