public void setPolicyManager(FederationPolicyManager policyManager)
      throws YarnException {
    if (policyManager == null) {
      LOG.warn("Attempting to set null policy manager");
      return;
    }
    // Extract the configuration from the policy manager
    String queue = policyManager.getQueue();
    SubClusterPolicyConfiguration conf;
    try {
      conf = policyManager.serializeConf();
    } catch (FederationPolicyInitializationException e) {
      LOG.warn("Error serializing policy for queue {}", queue);
      throw e;
    }
    if (conf == null) {
      // State store does not currently support setting a policy back to null
      // because it reads the queue name to set from the policy!
      LOG.warn("Skip setting policy to null for queue {} into state store",
          queue);
      return;
    }
    // Compare with configuration cache, if different, write the conf into
    // store and update our conf and manager cache
    if (!confCacheEqual(queue, conf)) {
      try {
        if (readOnly) {
          LOG.info("[read-only] Skipping policy update for queue {}", queue);
          return;
        }
        LOG.info("Updating policy for queue {} into state store", queue);
        stateStore.setPolicyConfiguration(conf);
        policyConfMap.put(queue, conf);
        policyManagerMap.put(queue, policyManager);
      } catch (YarnException e) {
        LOG.warn("Error writing SubClusterPolicyConfiguration to state "
            + "store for queue: {}", queue);
        throw e;
      }
    } else {
      LOG.info("Setting unchanged policy - state store write skipped");
    }
  }
@Test
  public void testReadOnly() throws YarnException {
    conf.setBoolean(YarnConfiguration.GPG_POLICY_GENERATOR_READONLY, true);
    stateStore = mock(MemoryFederationStateStore.class);
    facade.reinitialize(stateStore, conf);
    when(stateStore.getPolicyConfiguration(Matchers.any(
        GetSubClusterPolicyConfigurationRequest.class))).thenReturn(
        GetSubClusterPolicyConfigurationResponse.newInstance(testConf));
    policyFacade = new GPGPolicyFacade(facade, conf);

    // If we set a policy, the state store should be untouched
    WeightedLocalityPolicyManager manager =
        new WeightedLocalityPolicyManager();
    // Add a test policy for test queue
    manager.setQueue(TEST_QUEUE);
    manager.getWeightedPolicyInfo().setAMRMPolicyWeights(
        GPGUtils.createUniformWeights(subClusterIds));
    manager.getWeightedPolicyInfo().setRouterPolicyWeights(
        GPGUtils.createUniformWeights(subClusterIds));
    policyFacade.setPolicyManager(manager);
    verify(stateStore, times(0)).setPolicyConfiguration(
        Matchers.any(SetSubClusterPolicyConfigurationRequest.class));
  }