@Override
  public SaveFederationQueuePolicyResponse saveFederationQueuePolicy(
      SaveFederationQueuePolicyRequest request) throws YarnException, IOException {

    // Parameter validation.

    if (request == null) {
      routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
      RouterServerUtil.logAndThrowException("Missing SaveFederationQueuePolicy request.", null);
    }

    FederationQueueWeight federationQueueWeight = request.getFederationQueueWeight();
    if (federationQueueWeight == null) {
      routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
      RouterServerUtil.logAndThrowException("Missing FederationQueueWeight information.", null);
    }

    String queue = request.getQueue();
    if (StringUtils.isBlank(queue)) {
      routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
      RouterServerUtil.logAndThrowException("Missing Queue information.", null);
    }

    String policyManagerClassName = request.getPolicyManagerClassName();
    if (!checkPolicyManagerValid(policyManagerClassName, SUPPORT_WEIGHT_MANAGERS)) {
      routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
      RouterServerUtil.logAndThrowException(policyManagerClassName +
          " does not support the use of queue weights.", null);
    }

    String amRmWeight = federationQueueWeight.getAmrmWeight();
    FederationQueueWeight.checkSubClusterQueueWeightRatioValid(amRmWeight);

    String routerWeight = federationQueueWeight.getRouterWeight();
    FederationQueueWeight.checkSubClusterQueueWeightRatioValid(routerWeight);

    String headRoomAlpha = federationQueueWeight.getHeadRoomAlpha();
    FederationQueueWeight.checkHeadRoomAlphaValid(headRoomAlpha);

    try {
      long startTime = clock.getTime();

      // Step2, parse amRMPolicyWeights.
      Map<SubClusterIdInfo, Float> amRMPolicyWeights = getSubClusterWeightMap(amRmWeight);
      LOG.debug("amRMPolicyWeights = {}.", amRMPolicyWeights);

      // Step3, parse routerPolicyWeights.
      Map<SubClusterIdInfo, Float> routerPolicyWeights = getSubClusterWeightMap(routerWeight);
      LOG.debug("routerWeights = {}.", amRMPolicyWeights);

      // Step4, Initialize WeightedPolicyInfo.
      WeightedPolicyInfo weightedPolicyInfo = new WeightedPolicyInfo();
      weightedPolicyInfo.setHeadroomAlpha(Float.parseFloat(headRoomAlpha));
      weightedPolicyInfo.setAMRMPolicyWeights(amRMPolicyWeights);
      weightedPolicyInfo.setRouterPolicyWeights(routerPolicyWeights);

      // Step5, Set SubClusterPolicyConfiguration.
      SubClusterPolicyConfiguration policyConfiguration =
          SubClusterPolicyConfiguration.newInstance(queue, policyManagerClassName,
          weightedPolicyInfo.toByteBuffer());
      federationFacade.setPolicyConfiguration(policyConfiguration);
      long stopTime = clock.getTime();
      routerMetrics.succeededSaveFederationQueuePolicyRetrieved(stopTime - startTime);
      return SaveFederationQueuePolicyResponse.newInstance("save policy success.");
    } catch (Exception e) {
      routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
      RouterServerUtil.logAndThrowException(e,
          "Unable to saveFederationQueuePolicy due to exception. " + e.getMessage());
    }

    routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved();
    throw new YarnException("Unable to saveFederationQueuePolicy.");
  }
@Test
  public void testSaveFederationQueuePolicyErrorRequest() throws Exception {
    // null request.
    LambdaTestUtils.intercept(YarnException.class, "Missing SaveFederationQueuePolicy request.",
        () -> interceptor.saveFederationQueuePolicy(null));

    // federationQueueWeight is null.
    LambdaTestUtils.intercept(
        IllegalArgumentException.class, "FederationQueueWeight cannot be null.",
        () -> SaveFederationQueuePolicyRequest.newInstance("root.a", null, "-"));

    // queue is null
    FederationQueueWeight federationQueueWeight =
        FederationQueueWeight.newInstance("SC-1:0.7,SC-2:0.3", "SC-1:0.7,SC-2:0.3", "1.0");
    SaveFederationQueuePolicyRequest request =
        SaveFederationQueuePolicyRequest.newInstance("", federationQueueWeight, "-");
    LambdaTestUtils.intercept(YarnException.class, "Missing Queue information.",
        () -> interceptor.saveFederationQueuePolicy(request));

    // PolicyManager needs to support weight
    FederationQueueWeight federationQueueWeight2 = FederationQueueWeight.newInstance(
        "SC-1:0.7,SC-2:0.3", "SC-1:0.8,SC-2:0.3", "1.0");
    SaveFederationQueuePolicyRequest request2 =
        SaveFederationQueuePolicyRequest.newInstance("root.a", federationQueueWeight2,
        "TestPolicyManager");
    LambdaTestUtils.intercept(YarnException.class,
        "TestPolicyManager does not support the use of queue weights.",
        () -> interceptor.saveFederationQueuePolicy(request2));

    // routerWeight / amrmWeight
    // The sum of the routerWeight is not equal to 1.
    String policyTypeName = WeightedLocalityPolicyManager.class.getCanonicalName();
    FederationQueueWeight federationQueueWeight3 = FederationQueueWeight.newInstance(
        "SC-1:0.7,SC-2:0.3", "SC-1:0.8,SC-2:0.3", "1.0");
    SaveFederationQueuePolicyRequest request3 =
        SaveFederationQueuePolicyRequest.newInstance("root.a", federationQueueWeight3,
        policyTypeName);
    LambdaTestUtils.intercept(YarnException.class,
        "The sum of ratios for all subClusters must be equal to 1.",
        () -> interceptor.saveFederationQueuePolicy(request3));
  }