HDFS-11403. Zookeper ACLs on NN HA enabled clusters should be handled consistently. Contributed by Hanisha Koneru.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java

@@ -346,8 +346,13 @@ public synchronized void ensureParentZNode()
         createWithRetries(prefixPath, new byte[]{}, zkAcl, CreateMode.PERSISTENT);
       } catch (KeeperException e) {
         if (isNodeExists(e.code())) {
-          // This is OK - just ensuring existence.
-          continue;
+          // Set ACLs for parent node, if they do not exist or are different
+          try {
+            setAclsWithRetries(prefixPath);
+          } catch (KeeperException e1) {
+            throw new IOException("Couldn't set ACLs on parent ZNode: " +
+                prefixPath, e1);
+          }
         } else {
           throw new IOException("Couldn't create " + prefixPath, e);
         }
@@ -1066,14 +1071,36 @@ public Void run() throws KeeperException, InterruptedException {
     });
   }
 
+  private void setAclsWithRetries(final String path)
+      throws KeeperException, InterruptedException {
+    Stat stat = new Stat();
+    zkDoWithRetries(new ZKAction<Void>() {
+      @Override
+      public Void run() throws KeeperException, InterruptedException {
+        List<ACL> acl = zkClient.getACL(path, stat);
+        if (acl == null || !acl.containsAll(zkAcl) ||
+            !zkAcl.containsAll(acl)) {
+          zkClient.setACL(path, zkAcl, stat.getVersion());
+        }
+        return null;
+      }
+    }, Code.BADVERSION);
+  }
+
   private <T> T zkDoWithRetries(ZKAction<T> action) throws KeeperException,
       InterruptedException {
+    return zkDoWithRetries(action, null);
+  }
+
+  private <T> T zkDoWithRetries(ZKAction<T> action, Code retryCode)
+      throws KeeperException, InterruptedException {
     int retry = 0;
     while (true) {
       try {
         return action.run();
       } catch (KeeperException ke) {
-        if (shouldRetry(ke.code()) && ++retry < maxRetryNum) {
+        if ((shouldRetry(ke.code()) || shouldRetry(ke.code(), retryCode))
+            && ++retry < maxRetryNum) {
           continue;
         }
         throw ke;
@@ -1189,6 +1216,10 @@ private static boolean isSessionExpired(Code code) {
   private static boolean shouldRetry(Code code) {
     return code == Code.CONNECTIONLOSS || code == Code.OPERATIONTIMEOUT;
   }
+
+  private static boolean shouldRetry(Code code, Code retryIfCode) {
+    return (retryIfCode == null ? false : retryIfCode == code);
+  }
   
   @Override
   public String toString() {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java

@@ -715,6 +715,37 @@ public void testEnsureBaseNode() throws Exception {
       }
     }
   }
+
+  /**
+   * Test that ACLs are set on parent zNode even if the node already exists.
+   */
+  @Test
+  public void testParentZNodeACLs() throws Exception {
+    KeeperException ke = new KeeperException(Code.NODEEXISTS) {
+      @Override
+      public Code code() {
+        return super.code();
+      }
+    };
+
+    Mockito.when(mockZK.create(Mockito.anyString(), Mockito.eq(new byte[]{}),
+        Mockito.anyListOf(ACL.class),
+        Mockito.eq(CreateMode.PERSISTENT))).thenThrow(ke);
+
+    elector.ensureParentZNode();
+
+    StringBuilder prefix = new StringBuilder();
+    for (String part : ZK_PARENT_NAME.split("/")) {
+      if (part.isEmpty()) continue;
+      prefix.append("/").append(part);
+      if (!"/".equals(prefix.toString())) {
+        Mockito.verify(mockZK).getACL(Mockito.eq(prefix.toString()),
+            Mockito.eq(new Stat()));
+        Mockito.verify(mockZK).setACL(Mockito.eq(prefix.toString()),
+            Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.anyInt());
+      }
+    }
+  }
   
   /**
    * Test for a bug encountered during development of HADOOP-8163: