YARN-11435. [Router] FederationStateStoreFacade is not reinitialized with Router conf. (#5967) Contributed by Shilun Fan.

Reviewed-by: Inigo Goiri <inigoiri@apache.org>
Signed-off-by: Shilun Fan <slfan1989@apache.org>

Author: slfan1989

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestFederationRMFailoverProxyProvider.java

@@ -77,7 +77,7 @@ public void setUp() throws IOException, YarnException {
 
     stateStore = spy(new MemoryFederationStateStore());
     stateStore.init(conf);
-    FederationStateStoreFacade.getInstance().reinitialize(stateStore, conf);
+    FederationStateStoreFacade.getInstance(conf).reinitialize(stateStore, conf);
     verify(stateStore, times(0))
         .getSubClusters(any(GetSubClustersInfoRequest.class));
   }
@@ -180,7 +180,7 @@ public void run() {
           .getSubClusters(any(GetSubClustersInfoRequest.class));
 
       // Force flush cache, so that it will pick up the new RM address
-      FederationStateStoreFacade.getInstance().getSubCluster(subClusterId,
+      FederationStateStoreFacade.getInstance(conf).getSubCluster(subClusterId,
           true);
     }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/failover/FederationRMFailoverProxyProvider.java

@@ -76,7 +76,7 @@ public void init(Configuration configuration, RMProxy<T> proxy,
     String clusterId = configuration.get(YarnConfiguration.RM_CLUSTER_ID);
     Preconditions.checkNotNull(clusterId, "Missing RM ClusterId");
     this.subClusterId = SubClusterId.newInstance(clusterId);
-    this.facade = FederationStateStoreFacade.getInstance();
+    this.facade = FederationStateStoreFacade.getInstance(configuration);
     if (configuration instanceof YarnConfiguration) {
       this.conf = (YarnConfiguration) configuration;
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/utils/FederationStateStoreFacade.java

@@ -105,8 +105,7 @@ public final class FederationStateStoreFacade {
   private static final Logger LOG =
       LoggerFactory.getLogger(FederationStateStoreFacade.class);
 
-  private static final FederationStateStoreFacade FACADE =
-      new FederationStateStoreFacade();
+  private static volatile FederationStateStoreFacade facade;
 
   private static Random rand = new Random(System.currentTimeMillis());
 
@@ -115,8 +114,8 @@ public final class FederationStateStoreFacade {
   private SubClusterResolver subclusterResolver;
   private FederationCache federationCache;
 
-  private FederationStateStoreFacade() {
-    initializeFacadeInternal(new Configuration());
+  private FederationStateStoreFacade(Configuration conf) {
+    initializeFacadeInternal(conf);
   }
 
   private void initializeFacadeInternal(Configuration config) {
@@ -199,7 +198,50 @@ public static RetryPolicy createRetryPolicy(Configuration conf) {
    * @return the singleton {@link FederationStateStoreFacade} instance
    */
   public static FederationStateStoreFacade getInstance() {
-    return FACADE;
+    return getInstanceInternal(new Configuration());
+  }
+
+  /**
+   * Returns the singleton instance of the FederationStateStoreFacade object.
+   *
+   * @param conf configuration.
+   * @return the singleton {@link FederationStateStoreFacade} instance
+   */
+  public static FederationStateStoreFacade getInstance(Configuration conf) {
+    return getInstanceInternal(conf);
+  }
+
+  /**
+   * Returns the singleton instance of the FederationStateStoreFacade object.
+   *
+   * @param conf configuration.
+   * @return the singleton {@link FederationStateStoreFacade} instance
+   */
+  private static FederationStateStoreFacade getInstanceInternal(Configuration conf){
+    if (facade != null) {
+      return facade;
+    }
+    generateStateStoreFacade(conf);
+    return facade;
+  }
+
+  /**
+   * Generate the singleton instance of the FederationStateStoreFacade object.
+   *
+   * @param conf configuration.
+   */
+  private static void generateStateStoreFacade(Configuration conf){
+    if (facade == null) {
+      synchronized (FederationStateStoreFacade.class) {
+        if (facade == null) {
+          Configuration yarnConf = new Configuration();
+          if (conf != null) {
+            yarnConf = conf;
+          }
+          facade = new FederationStateStoreFacade(yarnConf);
+        }
+      }
+    }
   }
 
   /**

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/cache/TestFederationCache.java

@@ -60,13 +60,14 @@ public static Collection<Class[]> getParameters() {
   private Configuration conf;
   private FederationStateStore stateStore;
   private FederationStateStoreTestUtil stateStoreTestUtil;
-  private FederationStateStoreFacade facade = FederationStateStoreFacade.getInstance();
+  private FederationStateStoreFacade facade;
 
   public TestFederationCache(Class cacheClassName) {
     conf = new Configuration();
     conf.setInt(YarnConfiguration.FEDERATION_CACHE_TIME_TO_LIVE_SECS, 1);
     conf.setClass(YarnConfiguration.FEDERATION_FACADE_CACHE_CLASS,
         cacheClassName, FederationCache.class);
+    facade = FederationStateStoreFacade.getInstance(conf);
   }
 
   @Before

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/BaseFederationPoliciesTest.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.Random;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -75,7 +76,8 @@ public void testReinitilialize() throws YarnException {
         .newInstance("queue1", getPolicy().getClass().getCanonicalName(), buf));
     fpc.setFederationSubclusterResolver(
         FederationPoliciesTestUtil.initResolver());
-    fpc.setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade());
+    Configuration conf = new Configuration();
+    fpc.setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade(conf));
     getPolicy().reinitialize(fpc);
   }
 
@@ -100,7 +102,8 @@ public void testReinitilializeBad3() throws YarnException {
         .newInstance("queue1", "WrongPolicyName", buf));
     fpc.setFederationSubclusterResolver(
         FederationPoliciesTestUtil.initResolver());
-    fpc.setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade());
+    Configuration conf = new Configuration();
+    fpc.setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade(conf));
     getPolicy().reinitialize(fpc);
   }
 
@@ -212,9 +215,9 @@ public String generateClusterMetricsInfo(int id) {
   public FederationStateStoreFacade getMemoryFacade() throws YarnException {
 
     // setting up a store and its facade (with caching off)
-    FederationStateStoreFacade fedFacade = FederationStateStoreFacade.getInstance();
     YarnConfiguration conf = new YarnConfiguration();
     conf.setInt(YarnConfiguration.FEDERATION_CACHE_TIME_TO_LIVE_SECS, 0);
+    FederationStateStoreFacade fedFacade = FederationStateStoreFacade.getInstance(conf);
     FederationStateStore store = new MemoryFederationStateStore();
     store.init(conf);
     fedFacade.reinitialize(store, conf);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/TestFederationPolicyInitializationContextValidator.java

@@ -20,6 +20,7 @@
 
 import java.nio.ByteBuffer;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.FederationAMRMProxyPolicy;
 import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException;
 import org.apache.hadoop.yarn.server.federation.policies.manager.FederationPolicyManager;
@@ -45,7 +46,8 @@ public class TestFederationPolicyInitializationContextValidator {
 
   @Before
   public void setUp() throws Exception {
-    goodFacade = FederationPoliciesTestUtil.initFacade();
+    Configuration conf = new Configuration();
+    goodFacade = FederationPoliciesTestUtil.initFacade(conf);
     goodConfig = new MockPolicyManager().serializeConf();
     goodSR = FederationPoliciesTestUtil.initResolver();
     goodHome = SubClusterId.newInstance("homesubcluster");

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/TestRouterPolicyFacade.java

@@ -62,10 +62,9 @@ public class TestRouterPolicyFacade {
   public void setup() throws YarnException {
 
     // setting up a store and its facade (with caching off)
-    FederationStateStoreFacade fedFacade =
-        FederationStateStoreFacade.getInstance();
     YarnConfiguration conf = new YarnConfiguration();
     conf.set(YarnConfiguration.FEDERATION_CACHE_TIME_TO_LIVE_SECS, "0");
+    FederationStateStoreFacade fedFacade = FederationStateStoreFacade.getInstance(conf);
     store = new MemoryFederationStateStore();
     store.init(conf);
     fedFacade.reinitialize(store, conf);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/BasePolicyManagerTest.java

@@ -19,6 +19,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext;
 import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.FederationAMRMProxyPolicy;
 import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException;
@@ -72,13 +73,14 @@ protected static void serializeAndDeserializePolicyManager(
       Class expAMRMProxyPolicy, Class expRouterPolicy) throws Exception {
 
     // serializeConf it in a context
+    Configuration conf = new Configuration();
     SubClusterPolicyConfiguration fpc = wfp.serializeConf();
     fpc.setType(policyManagerType.getCanonicalName());
     FederationPolicyInitializationContext context =
         new FederationPolicyInitializationContext();
     context.setSubClusterPolicyConfiguration(fpc);
     context
-        .setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade());
+        .setFederationStateStoreFacade(FederationPoliciesTestUtil.initFacade(conf));
     context.setFederationSubclusterResolver(
         FederationPoliciesTestUtil.initResolver());
     context.setHomeSubcluster(SubClusterId.newInstance("homesubcluster"));

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java

@@ -123,7 +123,7 @@ public static void initializePolicyContext(
     fpc.setSubClusterPolicyConfiguration(SubClusterPolicyConfiguration
         .newInstance("queue1", policy.getClass().getCanonicalName(), buf));
     FederationStateStoreFacade facade = FederationStateStoreFacade
-        .getInstance();
+        .getInstance(conf);
     FederationStateStore fss = mock(FederationStateStore.class);
 
     if (activeSubclusters == null) {
@@ -242,9 +242,8 @@ public static SubClusterResolver initResolver() {
 
   public static FederationStateStoreFacade initFacade(
       List<SubClusterInfo> subClusterInfos, SubClusterPolicyConfiguration
-      policyConfiguration) throws YarnException {
-    FederationStateStoreFacade goodFacade = FederationStateStoreFacade
-        .getInstance();
+      policyConfiguration, Configuration conf) throws YarnException {
+    FederationStateStoreFacade goodFacade = FederationStateStoreFacade.getInstance(conf);
     FederationStateStore fss = mock(FederationStateStore.class);
     GetSubClustersInfoResponse response = GetSubClustersInfoResponse
         .newInstance(subClusterInfos);
@@ -276,8 +275,20 @@ public static FederationStateStoreFacade initFacade(
    * @throws YarnException in case the initialization is not successful.
    */
   public static FederationStateStoreFacade initFacade() throws YarnException {
+    return initFacade(new Configuration());
+  }
+
+  /**
+   * Initialiaze a main-memory {@link FederationStateStoreFacade} used for
+   * testing, wiht a mock resolver.
+   *
+   * @param conf Configuration.
+   * @return the facade.
+   * @throws YarnException in case the initialization is not successful.
+   */
+  public static FederationStateStoreFacade initFacade(Configuration conf) throws YarnException {
     SubClusterPolicyConfiguration policyConfiguration =
         SubClusterPolicyConfiguration.newInstance(null, null, null);
-    return initFacade(new ArrayList<>(), policyConfiguration);
+    return initFacade(new ArrayList<>(), policyConfiguration, conf);
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/TestFederationStateStoreFacade.java

@@ -80,8 +80,8 @@ public static Collection<Boolean[]> getParameters() {
   private Configuration conf;
   private FederationStateStore stateStore;
   private FederationStateStoreTestUtil stateStoreTestUtil;
-  private FederationStateStoreFacade facade =
-      FederationStateStoreFacade.getInstance();
+  private FederationStateStoreFacade facade;
+
   private Boolean isCachingEnabled;
 
   public TestFederationStateStoreFacade(Boolean isCachingEnabled) {
@@ -90,6 +90,7 @@ public TestFederationStateStoreFacade(Boolean isCachingEnabled) {
       conf.setInt(YarnConfiguration.FEDERATION_CACHE_TIME_TO_LIVE_SECS, 0);
     }
     this.isCachingEnabled = isCachingEnabled;
+    facade = FederationStateStoreFacade.getInstance(conf);
   }
 
   @Before