YARN-8085. ResourceProfilesManager should be set in RMActiveServiceContext. Contributed by Tao Yang.

(cherry picked from commit 7a59d60e0c50e8264e07f6b1dadaeedba676c8c2)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMActiveServiceContext.java

@@ -36,6 +36,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.placement.PlacementManager;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
 import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystem;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceProfilesManager;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.monitor.RMAppLifetimeMonitor;
@@ -111,6 +112,7 @@ public class RMActiveServiceContext {
   private QueueLimitCalculator queueLimitCalculator;
   private AllocationTagsManager allocationTagsManager;
   private PlacementConstraintManager placementConstraintManager;
+  private ResourceProfilesManager resourceProfilesManager;
 
   public RMActiveServiceContext() {
     queuePlacementManager = new PlacementManager();
@@ -513,4 +515,13 @@ public class RMActiveServiceContext {
       QueueLimitCalculator limitCalculator) {
     this.queueLimitCalculator = limitCalculator;
   }
+
+  public ResourceProfilesManager getResourceProfilesManager() {
+    return resourceProfilesManager;
+  }
+
+  public void setResourceProfilesManager(
+      ResourceProfilesManager resourceProfilesManager) {
+    this.resourceProfilesManager = resourceProfilesManager;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java

@@ -93,8 +93,6 @@ public class RMContextImpl implements RMContext {
    */
   private RMActiveServiceContext activeServiceContext;
 
-  private ResourceProfilesManager resourceProfilesManager;
-
   private String proxyHostAndPort = null;
 
   /**
@@ -591,7 +589,7 @@ public class RMContextImpl implements RMContext {
 
   @Override
   public ResourceProfilesManager getResourceProfilesManager() {
-    return this.resourceProfilesManager;
+    return this.activeServiceContext.getResourceProfilesManager();
   }
 
   String getProxyHostAndPort(Configuration conf) {
@@ -619,7 +617,7 @@ public class RMContextImpl implements RMContext {
 
   @Override
   public void setResourceProfilesManager(ResourceProfilesManager mgr) {
-    this.resourceProfilesManager = mgr;
+    this.activeServiceContext.setResourceProfilesManager(mgr);
   }
   // Note: Read java doc before adding any services over here.
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java

@@ -245,13 +245,6 @@ public class ResourceManager extends CompositeService implements Recoverable {
     this.rmContext = new RMContextImpl();
     rmContext.setResourceManager(this);
 
-
-    // add resource profiles here because it's used by AbstractYarnScheduler
-    ResourceProfilesManager resourceProfilesManager =
-        createResourceProfileManager();
-    resourceProfilesManager.init(conf);
-    rmContext.setResourceProfilesManager(resourceProfilesManager);
-
     this.configurationProvider =
         ConfigurationProviderFactory.getConfigurationProvider(conf);
     this.configurationProvider.init(this.conf);
@@ -641,6 +634,12 @@ public class ResourceManager extends CompositeService implements Recoverable {
       addService(placementConstraintManager);
       rmContext.setPlacementConstraintManager(placementConstraintManager);
 
+      // add resource profiles here because it's used by AbstractYarnScheduler
+      ResourceProfilesManager resourceProfilesManager =
+          createResourceProfileManager();
+      resourceProfilesManager.init(conf);
+      rmContext.setResourceProfilesManager(resourceProfilesManager);
+
       RMDelegatedNodeLabelsUpdater delegatedNodeLabelsUpdater =
           createRMDelegatedNodeLabelsUpdater();
       if (delegatedNodeLabelsUpdater != null) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMHA.java

@@ -658,6 +658,45 @@ public class TestRMHA {
     assertEquals(HAServiceState.STANDBY, rm.getRMContext().getHAServiceState());
   }
 
+  @Test
+  public void testResourceProfilesManagerAfterRMWentStandbyThenBackToActive()
+      throws Exception {
+    configuration.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false);
+    configuration.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
+    Configuration conf = new YarnConfiguration(configuration);
+    conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
+
+    // 1. start RM
+    rm = new MockRM(conf);
+    rm.init(conf);
+    rm.start();
+
+    StateChangeRequestInfo requestInfo = new StateChangeRequestInfo(
+        HAServiceProtocol.RequestSource.REQUEST_BY_USER);
+    checkMonitorHealth();
+    checkStandbyRMFunctionality();
+
+    // 2. Transition to active
+    rm.adminService.transitionToActive(requestInfo);
+    checkMonitorHealth();
+    checkActiveRMFunctionality();
+
+    // 3. Transition to standby
+    rm.adminService.transitionToStandby(requestInfo);
+    checkMonitorHealth();
+    checkStandbyRMFunctionality();
+
+    // 4. Transition to active
+    rm.adminService.transitionToActive(requestInfo);
+    checkMonitorHealth();
+    checkActiveRMFunctionality();
+
+    // 5. Check ResourceProfilesManager
+    Assert.assertNotNull(
+        "ResourceProfilesManager should not be null!",
+        rm.getRMContext().getResourceProfilesManager());
+  }
+
   public void innerTestHAWithRMHostName(boolean includeBindHost) {
     //this is run two times, with and without a bind host configured
     if (includeBindHost) {