public FederationPolicyManager getPolicyManager(String queueName)
      throws YarnException {
    FederationPolicyManager policyManager = policyManagerMap.get(queueName);

    // If we don't have the policy manager cached, pull configuration
    // from the FederationStateStore to create and cache it
    if (policyManager == null) {
      try {

        // If we don't have the configuration cached, pull it
        // from the stateStore
        SubClusterPolicyConfiguration conf = policyConfMap.get(queueName);

        if (conf == null) {
          conf = stateStore.getPolicyConfiguration(queueName);
        }

        // If configuration is still null, it does not exist in the
        // FederationStateStore
        if (conf == null) {
          LOG.info("Read null policy for queue {}.", queueName);
          return null;
        }

        // Generate PolicyManager based on PolicyManagerType.
        String policyManagerType = conf.getType();
        policyManager = FederationPolicyUtils.instantiatePolicyManager(policyManagerType);
        policyManager.setQueue(queueName);

        // If PolicyManager supports Weighted PolicyInfo, it means that
        // we need to use this parameter to determine which sub-cluster the router goes to
        // or which sub-cluster the container goes to.
        if (policyManager.isSupportWeightedPolicyInfo()) {
          ByteBuffer weightedPolicyInfoParams = conf.getParams();
          if (weightedPolicyInfoParams == null) {
            LOG.warn("Warning: Queue = {}, FederationPolicyManager {} WeightedPolicyInfo is empty.",
                queueName, policyManagerType);
            return null;
          }
          WeightedPolicyInfo weightedPolicyInfo =
              WeightedPolicyInfo.fromByteBuffer(conf.getParams());
          policyManager.setWeightedPolicyInfo(weightedPolicyInfo);
        } else {
          LOG.warn("Warning: FederationPolicyManager of unsupported WeightedPolicyInfo type {}, " +
              "initialization may be incomplete.", policyManager.getClass());
        }

        policyManagerMap.put(queueName, policyManager);
        policyConfMap.put(queueName, conf);
      } catch (YarnException e) {
        LOG.error("Error reading SubClusterPolicyConfiguration from state "
            + "store for queue: {}", queueName);
        throw e;
      }
    }
    return policyManager;
  }
@Test
  public void testGetWeightedHomePolicyManager() throws YarnException {
    stateStore = new MemoryFederationStateStore();
    stateStore.init(new Configuration());

    // root.b uses WeightedHomePolicyManager.
    // Step1. Prepare routerPolicyWeights.
    Map<SubClusterIdInfo, Float> routerPolicyWeights = new HashMap<>();
    routerPolicyWeights.put(new SubClusterIdInfo("SC-1"), 0.8f);
    routerPolicyWeights.put(new SubClusterIdInfo("SC-2"), 0.2f);

    WeightedPolicyInfo weightedPolicyInfo = new WeightedPolicyInfo();
    weightedPolicyInfo.setHeadroomAlpha(1);
    weightedPolicyInfo.setRouterPolicyWeights(routerPolicyWeights);

    // Step2. Set PolicyConfiguration.
    String policyManagerType = WeightedHomePolicyManager.class.getName();
    SubClusterPolicyConfiguration config = SubClusterPolicyConfiguration.newInstance("root.b",
        policyManagerType, weightedPolicyInfo.toByteBuffer());
    SetSubClusterPolicyConfigurationRequest request =
        SetSubClusterPolicyConfigurationRequest.newInstance(config);
    stateStore.setPolicyConfiguration(request);

    // Step3. Get FederationPolicyManager using policyFacade.
    facade.reinitialize(stateStore, conf);
    policyFacade = new GPGPolicyFacade(facade, conf);
    FederationPolicyManager policyManager = policyFacade.getPolicyManager("root.b");
    Assert.assertNotNull(policyManager);
    Assert.assertTrue(policyManager.isSupportWeightedPolicyInfo());
    WeightedPolicyInfo weightedPolicyInfo1 = policyManager.getWeightedPolicyInfo();
    Assert.assertNotNull(weightedPolicyInfo1);

    // Step4. Confirm amrmPolicyWeight is accurate.
    Map<SubClusterIdInfo, Float> amrmPolicyWeights1 = weightedPolicyInfo1.getAMRMPolicyWeights();
    Assert.assertNotNull(amrmPolicyWeights1);
    Assert.assertEquals(0, amrmPolicyWeights1.size());

    // Step5. Confirm amrmPolicyWeight is accurate.
    Map<SubClusterIdInfo, Float> routerPolicyWeights1 =
        weightedPolicyInfo1.getRouterPolicyWeights();
    Assert.assertNotNull(routerPolicyWeights1);
    Float sc1Float1 = routerPolicyWeights1.get(new SubClusterIdInfo("SC-1"));
    Float sc2Float2 = routerPolicyWeights1.get(new SubClusterIdInfo("SC-2"));
    Assert.assertEquals(0.8, sc1Float1, 0.001);
    Assert.assertEquals(0.2, sc2Float2, 0.001);
  }