YARN-11536. Fix CheckStyle.

Author: slfan1989

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/BatchSaveFederationQueuePoliciesResponse.java

@@ -0,0 +1,47 @@
+/**
+ * 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.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.util.Records;
+
+@Private
+@Unstable
+public abstract class BatchSaveFederationQueuePoliciesResponse {
+
+  public static BatchSaveFederationQueuePoliciesResponse newInstance() {
+    return Records.newRecord(BatchSaveFederationQueuePoliciesResponse.class);
+  }
+
+  public static BatchSaveFederationQueuePoliciesResponse newInstance(String msg) {
+    BatchSaveFederationQueuePoliciesResponse response =
+        Records.newRecord(BatchSaveFederationQueuePoliciesResponse.class);
+    response.setMessage(msg);
+    return response;
+  }
+
+  @Public
+  @Unstable
+  public abstract String getMessage();
+
+  @Public
+  @Unstable
+  public abstract void setMessage(String msg);
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/server/resourcemanager_administration_protocol.proto

@@ -49,4 +49,5 @@ service ResourceManagerAdministrationProtocolService {
   rpc mapAttributesToNodes(NodesToAttributesMappingRequestProto) returns (NodesToAttributesMappingResponseProto);
   rpc deregisterSubCluster(DeregisterSubClusterRequestProto) returns (DeregisterSubClusterResponseProto);
   rpc saveFederationQueuePolicy(SaveFederationQueuePolicyRequestProto) returns (SaveFederationQueuePolicyResponseProto);
+  rpc batchSaveFederationQueuePolicies(BatchSaveFederationQueuePoliciesRequestProto) returns (BatchSaveFederationQueuePoliciesResponseProto);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RouterCLI.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.hadoop.yarn.client.ClientRMProxy;
 import org.apache.hadoop.yarn.client.util.FormattingCLIUtils;
+import org.apache.hadoop.yarn.client.util.MemoryPageUtils;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.api.ResourceManagerAdministrationProtocol;
@@ -113,21 +114,20 @@ public class RouterCLI extends Configured implements Tool {
   // Command2: policy
   // save policy
   private static final String OPTION_S = "s";
-
   private static final String OPTION_BATCH_S = "bs";
-
   private static final String OPTION_SAVE = "save";
-
   private static final String OPTION_BATCH_SAVE = "batch-save";
-
   private static final String OPTION_FORMAT = "format";
+  private static final String OPTION_FILE = "f";
+  private static final String OPTION_INPUT_FILE = "input-file";
 
   private static final String CMD_POLICY = "-policy";
   private static final String FORMAT_XML = "xml";
   private static final String FORMAT_JSON = "json";
   private static final String XML_TAG_SUBCLUSTERIDINFO = "subClusterIdInfo";
   private static final String XML_TAG_AMRMPOLICYWEIGHTS = "amrmPolicyWeights";
   private static final String XML_TAG_ROUTERPOLICYWEIGHTS = "routerPolicyWeights";
+  private static final String XML_TAG_HEADROOMALPHA = "headroomAlpha";
 
   public RouterCLI() {
     super();
@@ -338,6 +338,10 @@ private int handlePolicy(String[] args)
         "Users can specify the file format using this option. " +
         "Currently, there are two supported file formats: XML and JSON. " +
         "These files contain the policy information for storing queue policies.");
+    Option fileOpt = new Option(null, "file", true,
+        "Users can specify the file format using this option. " +
+        "Currently, there are two supported file formats: XML and JSON. " +
+        "These files contain the policy information for storing queue policies.");
     formatOpt.setOptionalArg(true);
 
     opts.addOption(saveOpt);
@@ -366,16 +370,31 @@ private int handlePolicy(String[] args)
 
     // Save Queue Policies in Batches
     if (cliParser.hasOption(OPTION_BATCH_S) || cliParser.hasOption(OPTION_BATCH_SAVE)) {
+
+      // Determine whether the file format is accurate, XML or JSON format.
+      // If it is not XML or JSON, we will directly prompt the user with an error message.
+      String format = null;
       if (cliParser.hasOption(OPTION_FORMAT)) {
-        String format = cliParser.getOptionValue(OPTION_FORMAT);
+        format = cliParser.getOptionValue(OPTION_FORMAT);
         if (StringUtils.isBlank(format) ||
             StringUtils.equalsAnyIgnoreCase(format, FORMAT_XML, FORMAT_JSON)) {
           System.out.println("We currently only support policy configuration files " +
               "in XML and JSON formats.");
           return EXIT_ERROR;
         }
-        System.out.println("format:" + format);
       }
+
+      // Parse configuration file path
+      String filePath = null;
+      if (cliParser.hasOption(OPTION_FILE) || cliParser.hasOption(OPTION_INPUT_FILE)) {
+        filePath = cliParser.getOptionValue(OPTION_FILE);
+        if (StringUtils.isBlank(filePath)) {
+          filePath = cliParser.getOptionValue(OPTION_INPUT_FILE);
+        }
+      }
+
+      handBatchSavePolicies(format, filePath);
+
       System.out.println("We need to specify the format of the input file.");
     }
 
@@ -399,7 +418,16 @@ private int handleSavePolicy(String policy) {
   private int handBatchSavePolicies(String format, String policyFile) {
     LOG.info("Batch Save Federation Policies. Format = {}, PolicyFile = {}.",
         format, policyFile);
-    return 1;
+    switch (format) {
+      case FORMAT_JSON:
+        break;
+      case FORMAT_XML:
+        return parseXml2Policies(policyFile);
+      default:
+        System.out.println("We currently only support JSON and XML formats.");
+        return EXIT_ERROR;
+    }
+    return EXIT_ERROR;
   }
 
   /**
@@ -444,6 +472,22 @@ protected SaveFederationQueuePolicyRequest parsePolicy(String policy) throws Yar
     return request;
   }
 
+  protected int parseXml2Policies(String policiesXml) {
+    try {
+      List<FederationQueueWeight> federationQueueWeights = parsePoliciesByXml(policiesXml);
+      MemoryPageUtils<FederationQueueWeight> memoryPageUtils = new MemoryPageUtils<>(20);
+      federationQueueWeights.forEach(federationQueueWeight ->
+          memoryPageUtils.addToMemory(federationQueueWeight));
+      int pages = memoryPageUtils.getPages();
+      for (int i = 0; i < pages; i++) {
+
+      }
+    } catch (Exception e) {
+
+    }
+    return 0;
+  }
+
   protected List<FederationQueueWeight> parsePoliciesByXml(String policiesXml)
       throws IOException, SAXException, ParserConfigurationException {
 
@@ -474,10 +518,11 @@ protected List<FederationQueueWeight> parsePoliciesByXml(String policiesXml)
 
             // parse amrmPolicyWeights / routerPolicyWeights.
             String amrmWeight = parsePolicyWeightsNode(queueElement, XML_TAG_AMRMPOLICYWEIGHTS);
-            String routerWeight = parsePolicyWeightsNode(queueElement, "routerPolicyWeights");
+            String routerWeight = parsePolicyWeightsNode(queueElement, XML_TAG_ROUTERPOLICYWEIGHTS);
 
             // parse headroomAlpha.
-            String headroomAlpha = queueElement.getElementsByTagName("headroomAlpha").item(0).getTextContent();
+            String headroomAlpha = queueElement.getElementsByTagName(XML_TAG_HEADROOMALPHA)
+                .item(0).getTextContent();
 
             String policyManager = getConf().get(YarnConfiguration.FEDERATION_POLICY_MANAGER,
                 YarnConfiguration.DEFAULT_FEDERATION_POLICY_MANAGER);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/util/MemoryPageUtils.java

@@ -51,4 +51,8 @@ public List<T> readFromMemory(int pageNumber) {
     }
     return dataList.subList(startIndex, endIndex);
   }
+
+  public int getPages() {
+    return (dataList.size() / pageSize + 1);
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestRouterCLI.java

@@ -221,7 +221,7 @@ public void testSavePolicy() throws Exception {
   }
 
   @Test
-  public void testSavePolicies() throws Exception {
+  public void testParsePoliciesByXml() throws Exception {
     String filePath =
         TestRouterCLI.class.getClassLoader().getResource("federation-weights.xml").getFile();
     List<FederationQueueWeight> federationQueueWeights = rmAdminCLI.parsePoliciesByXml(filePath);
@@ -231,10 +231,15 @@ public void testSavePolicies() throws Exception {
     // Queue1: root.a
     FederationQueueWeight queueWeight1 = federationQueueWeights.get(0);
     assertNotNull(queueWeight1);
-    assertEquals("", queueWeight1);
+    assertEquals("root.a", queueWeight1.getQueue());
+    assertEquals("SC-1:0.7,SC-2:0.3", queueWeight1.getAmrmWeight());
+    assertEquals("SC-1:0.6,SC-2:0.4", queueWeight1.getRouterWeight());
 
     // Queue2: root.b
     FederationQueueWeight queueWeight2 = federationQueueWeights.get(1);
     assertNotNull(queueWeight2);
+    assertEquals("root.b", queueWeight2.getQueue());
+    assertEquals("SC-1:0.8,SC-2:0.2", queueWeight2.getAmrmWeight());
+    assertEquals("SC-1:0.6,SC-2:0.4", queueWeight2.getRouterWeight());
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/ResourceManagerAdministrationProtocolPBServiceImpl.java

@@ -22,6 +22,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.AddToClusterNodeLabelsRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.AddToClusterNodeLabelsResponseProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.CheckForDecommissioningNodesRequestProto;
@@ -399,4 +400,9 @@ public SaveFederationQueuePolicyResponseProto saveFederationQueuePolicy(RpcContr
       throw new ServiceException(e);
     }
   }
+
+  @Override
+  public YarnServerResourceManagerServiceProtos.BatchSaveFederationQueuePoliciesResponseProto batchSaveFederationQueuePolicies(RpcController controller, YarnServerResourceManagerServiceProtos.BatchSaveFederationQueuePoliciesRequestProto request) throws ServiceException {
+    return null;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/BatchSaveFederationQueuePoliciesRequestPBImpl.java

@@ -0,0 +1,146 @@
+/**
+ * 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.api.protocolrecords.impl.pb;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.thirdparty.protobuf.TextFormat;
+import org.apache.hadoop.yarn.proto.YarnProtos.FederationQueueWeightProto;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.BatchSaveFederationQueuePoliciesRequestProtoOrBuilder;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.BatchSaveFederationQueuePoliciesRequestProto;
+import org.apache.hadoop.yarn.server.api.protocolrecords.BatchSaveFederationQueuePoliciesRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.FederationQueueWeight;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Private
+@Unstable
+public class BatchSaveFederationQueuePoliciesRequestPBImpl
+    extends BatchSaveFederationQueuePoliciesRequest {
+
+  private BatchSaveFederationQueuePoliciesRequestProto proto =
+      BatchSaveFederationQueuePoliciesRequestProto.getDefaultInstance();
+  private BatchSaveFederationQueuePoliciesRequestProto.Builder builder = null;
+  private boolean viaProto = false;
+  private List<FederationQueueWeight> federationQueueWeights = null;
+
+  public BatchSaveFederationQueuePoliciesRequestPBImpl() {
+    this.builder = BatchSaveFederationQueuePoliciesRequestProto.newBuilder();
+  }
+
+  public BatchSaveFederationQueuePoliciesRequestPBImpl(
+      BatchSaveFederationQueuePoliciesRequestProto proto) {
+    this.proto = proto;
+    viaProto = true;
+  }
+
+  private synchronized void maybeInitBuilder() {
+    if (viaProto || builder == null) {
+      builder = BatchSaveFederationQueuePoliciesRequestProto.newBuilder(proto);
+    }
+    viaProto = false;
+  }
+
+  private void mergeLocalToProto() {
+    if (viaProto) {
+      maybeInitBuilder();
+    }
+    if (this.federationQueueWeights != null) {
+      for (FederationQueueWeight federationQueueWeight : federationQueueWeights) {
+        FederationQueueWeightPBImpl federationQueueWeightPBImpl =
+          (FederationQueueWeightPBImpl) federationQueueWeight;
+        builder.addFederationQueueWeights(federationQueueWeightPBImpl.getProto());
+      }
+    }
+    proto = builder.build();
+    viaProto = true;
+  }
+
+  public BatchSaveFederationQueuePoliciesRequestProto getProto() {
+    mergeLocalToProto();
+    proto = viaProto ? proto : builder.build();
+    viaProto = true;
+    return proto;
+  }
+
+  private void initDeregisterSubClustersMapping() {
+    if (this.federationQueueWeights != null) {
+        return;
+    }
+
+    BatchSaveFederationQueuePoliciesRequestProtoOrBuilder p = viaProto ? proto : builder;
+    List<FederationQueueWeightProto> batchSaveFederationQueuePoliciesProtoList =
+        p.getFederationQueueWeightsList();
+
+    List<FederationQueueWeight> attributes = new ArrayList<>();
+    if (batchSaveFederationQueuePoliciesProtoList == null ||
+        batchSaveFederationQueuePoliciesProtoList.size() == 0) {
+      this.federationQueueWeights = attributes;
+      return;
+    }
+
+    for (FederationQueueWeightProto federationQueueWeightProto :
+        batchSaveFederationQueuePoliciesProtoList) {
+      attributes.add(new FederationQueueWeightPBImpl(federationQueueWeightProto));
+    }
+
+    this.federationQueueWeights = attributes;
+}
+
+  @Override
+  public List<FederationQueueWeight> getFederationQueueWeights() {
+    initDeregisterSubClustersMapping();
+    return this.federationQueueWeights;
+  }
+
+  @Override
+  public void setFederationQueueWeights(List<FederationQueueWeight> pFederationQueueWeights) {
+    if (federationQueueWeights == null) {
+      federationQueueWeights = new ArrayList<>();
+    }
+    if(federationQueueWeights == null) {
+      throw new IllegalArgumentException("federationQueueWeights cannot be null");
+    }
+    federationQueueWeights.clear();
+    federationQueueWeights.addAll(pFederationQueueWeights);
+  }
+
+  @Override
+  public int hashCode() {
+    return getProto().hashCode();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (!(other instanceof BatchSaveFederationQueuePoliciesRequest)) {
+        return false;
+    }
+
+    BatchSaveFederationQueuePoliciesRequestPBImpl otherImpl = this.getClass().cast(other);
+    return new EqualsBuilder()
+        .append(this.getProto(), otherImpl.getProto())
+        .isEquals();
+  }
+
+  @Override
+  public String toString() {
+    return TextFormat.shortDebugString(getProto());
+  }
+}