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 testGetPolicy() throws YarnException {
    WeightedLocalityPolicyManager manager =
        (WeightedLocalityPolicyManager) policyFacade
            .getPolicyManager(TEST_QUEUE);
    Assert.assertEquals(testConf, manager.serializeConf());
  }