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 testGetUniformBroadcastPolicyManager() throws Exception {
    stateStore = new MemoryFederationStateStore();
    stateStore.init(new Configuration());

    List<String> notSupportWeightedPolicyInfos = new ArrayList<>();
    notSupportWeightedPolicyInfos.add(HashBroadcastPolicyManager.class.getName());
    notSupportWeightedPolicyInfos.add(UniformBroadcastPolicyManager.class.getName());
    notSupportWeightedPolicyInfos.add(HomePolicyManager.class.getName());
    notSupportWeightedPolicyInfos.add(RejectAllPolicyManager.class.getName());
    String prefix = "org.apache.hadoop.yarn.server.federation.policies.manager.";

    for (String policyManagerType : notSupportWeightedPolicyInfos) {
      // root.c uses UniformBroadcastPolicyManager.
      // Step1. Prepare routerPolicyWeights.
      WeightedPolicyInfo weightedPolicyInfo = new WeightedPolicyInfo();
      weightedPolicyInfo.setHeadroomAlpha(1);

      // Step2. Set PolicyConfiguration.
      SubClusterPolicyConfiguration config = SubClusterPolicyConfiguration.newInstance("root.c",
          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.c");
      Assert.assertNotNull(policyManager);
      Assert.assertFalse(policyManager.isSupportWeightedPolicyInfo());
      String policyManagerTypeSimple = policyManagerType.replace(prefix, "");
      // Verify that PolicyManager is initialized successfully,
      // but getWeightedPolicyInfo is not supported.
      LambdaTestUtils.intercept(NotImplementedException.class,
          policyManagerTypeSimple + " does not implement getWeightedPolicyInfo.",
          () -> policyManager.getWeightedPolicyInfo());
    }
  }