YARN-7453. Fix issue where RM fails to switch to active after first successful start. (Rohith Sharma K S via asuresh)

(cherry picked from commit a9c70b0e84dab0c41e480a0dc0cb1a22efdc64ee)

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

@@ -342,7 +342,7 @@ public class ResourceManager extends CompositeService implements Recoverable {
         conf.getBoolean(YarnConfiguration.CURATOR_LEADER_ELECTOR,
             YarnConfiguration.DEFAULT_CURATOR_LEADER_ELECTOR_ENABLED);
     if (curatorEnabled) {
-      this.zkManager = getAndStartZKManager(conf);
+      this.zkManager = createAndStartZKManager(conf);
       elector = new CuratorBasedElectorService(this);
     } else {
       elector = new ActiveStandbyElectorBasedElectorService(this);
@@ -356,11 +356,8 @@ public class ResourceManager extends CompositeService implements Recoverable {
    * @return ZooKeeper Curator manager.
    * @throws IOException If it cannot create the manager.
    */
-  public synchronized ZKCuratorManager getAndStartZKManager(Configuration
+  public ZKCuratorManager createAndStartZKManager(Configuration
       config) throws IOException {
-    if (this.zkManager != null) {
-      return zkManager;
-    }
     ZKCuratorManager manager = new ZKCuratorManager(config);
 
     // Get authentication
@@ -380,7 +377,10 @@ public class ResourceManager extends CompositeService implements Recoverable {
     }
 
     manager.start(authInfos);
-    this.zkManager = manager;
+    return manager;
+  }
+
+  public ZKCuratorManager getZKManager() {
     return zkManager;
   }
 

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

@@ -358,7 +358,10 @@ public class ZKRMStateStore extends RMStateStore {
     amrmTokenSecretManagerRoot =
         getNodePath(zkRootNodePath, AMRMTOKEN_SECRET_MANAGER_ROOT);
     reservationRoot = getNodePath(zkRootNodePath, RESERVATION_SYSTEM_ROOT);
-    zkManager = resourceManager.getAndStartZKManager(conf);
+    zkManager = resourceManager.getZKManager();
+    if(zkManager==null) {
+      zkManager = resourceManager.createAndStartZKManager(conf);
+    }
     delegationTokenNodeSplitIndex =
         conf.getInt(YarnConfiguration.ZK_DELEGATION_TOKEN_NODE_SPLIT_INDEX,
             YarnConfiguration.DEFAULT_ZK_DELEGATION_TOKEN_NODE_SPLIT_INDEX);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ZKConfigurationStore.java

@@ -81,7 +81,8 @@ public class ZKConfigurationStore extends YarnConfigurationStore {
     this.znodeParentPath =
         conf.get(YarnConfiguration.RM_SCHEDCONF_STORE_ZK_PARENT_PATH,
             YarnConfiguration.DEFAULT_RM_SCHEDCONF_STORE_ZK_PARENT_PATH);
-    this.zkManager = rmContext.getResourceManager().getAndStartZKManager(conf);
+    this.zkManager =
+        rmContext.getResourceManager().createAndStartZKManager(conf);
     this.zkAcl = ZKCuratorManager.getZKAcls(conf);
 
     this.zkVersionPath = getNodePath(znodeParentPath, ZK_VERSION_PATH);