CURATOR-728: Not issue ZooKeeper::create if possible in ZkPaths::mkdirs

`ZkPaths::mkdir("/bar/foo")` will not run into `NoAuthException` if
"/bar/foo" exists, and we have `READ` permission to "/bar/foo" but not
`CREATE` permission to "/bar".

Author: kezhuw

curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java

@@ -282,28 +282,29 @@ public static void mkdirs(ZooKeeper zookeeper, String path, boolean makeLastNode
      */
     public static void mkdirs(
             ZooKeeper zookeeper,
-            String path,
+            final String path,
             boolean makeLastNode,
             InternalACLProvider aclProvider,
             boolean asContainers)
             throws InterruptedException, KeeperException {
         PathUtils.validatePath(path);
 
-        int pos = path.length();
-        String subPath;
-
-        // This first loop locates the first parent that doesn't exist from leaf nodes towards root
+        // This first loop locates the first node that doesn't exist from leaf nodes towards root
         // this way, it is not required to have read access on all parents.
         // This is relevant after https://issues.apache.org/jira/browse/ZOOKEEPER-2590.
 
+        int pos = path.length();
         do {
+            String subPath = path.substring(0, pos);
+            if (zookeeper.exists(subPath, false) != null) {
+                break;
+            }
             pos = path.lastIndexOf(PATH_SEPARATOR_CHAR, pos - 1);
-            subPath = path.substring(0, pos);
-        } while (pos > 0 && zookeeper.exists(subPath, false) == null);
+        } while (pos > 0);
 
         // Start creating the subtree after the longest path that exists, assuming root always exists.
 
-        do {
+        while (pos < path.length()) {
             pos = path.indexOf(PATH_SEPARATOR_CHAR, pos + 1);
 
             if (pos == -1) {
@@ -314,7 +315,7 @@ public static void mkdirs(
                 }
             }
 
-            subPath = path.substring(0, pos);
+            String subPath = path.substring(0, pos);
 
             // All the paths from the initial `pos` do not exist.
             try {
@@ -332,8 +333,8 @@ public static void mkdirs(
             } catch (KeeperException.NodeExistsException ignore) {
                 // ignore... someone else has created it since we checked
             }
-
-        } while (pos < path.length());
+        }
+        ;
     }
 
     /**

curator-framework/src/test/java/org/apache/curator/framework/imps/TestCreate.java

@@ -691,14 +691,15 @@ public void testForbiddenAncestors() throws Exception {
 
             client.create().creatingParentsIfNeeded().forPath("/bat/bi/hiru");
             client.setACL()
-                    .withACL(Collections.singletonList(new ACL(0, ANYONE_ID_UNSAFE)))
+                    .withACL(Collections.singletonList(new ACL(ZooDefs.Perms.CREATE, ANYONE_ID_UNSAFE)))
                     .forPath("/bat");
 
             // In creation attempts where the parent ("/bat") has ACL that restricts read, creation request fails.
             try {
-                client.create().creatingParentsIfNeeded().forPath("/bat/bost");
+                client.create().creatingParentsIfNeeded().forPath("/bat/foo/bost");
                 fail("Expected NoAuthException when not authorized to read new node grandparent");
             } catch (KeeperException.NoAuthException noAuthException) {
+                assertEquals(noAuthException.getPath(), "/bat");
             }
 
             // But creating a node in the same subtree where its grandparent has read access is allowed and

curator-framework/src/test/java/org/apache/curator/framework/imps/TestEnsureContainers.java

@@ -19,14 +19,20 @@
 
 package org.apache.curator.framework.imps;
 
+import static org.apache.zookeeper.ZooDefs.Ids.ANYONE_ID_UNSAFE;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import java.util.Collections;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.EnsureContainers;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.utils.CloseableUtils;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
 import org.junit.jupiter.api.Test;
 
 public class TestEnsureContainers extends BaseClassForTests {
@@ -63,4 +69,31 @@ public void testSingleExecution() throws Exception {
             CloseableUtils.closeQuietly(client);
         }
     }
+
+    @Test
+    public void testNodeExistsButNoCreatePermission() throws Exception {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try {
+            client.start();
+
+            // given: "/bar/foo" created
+            client.create().creatingParentsIfNeeded().forPath("/bar/foo");
+            // given: only permission read to "/bar"
+            client.setACL()
+                    .withACL(Collections.singletonList(new ACL(ZooDefs.Perms.READ, ANYONE_ID_UNSAFE)))
+                    .forPath("/bar");
+
+            // check: create "/bar/foo" will fail with NoAuth
+            assertThrows(KeeperException.NoAuthException.class, () -> {
+                client.create().forPath("/bar/foo");
+            });
+
+            // when: mkdirs("/bar/foo")
+            // then: everything fine as "/bar/foo" exists, and we have READ permission
+            EnsureContainers ensureContainers = new EnsureContainers(client, "/bar/foo");
+            ensureContainers.ensure();
+        } finally {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
 }

curator-test-zk37/pom.xml

@@ -31,7 +31,7 @@
     <artifactId>curator-test-zk37</artifactId>
 
     <properties>
-        <zookeeper-37-version>3.7.2</zookeeper-37-version>
+        <zookeeper-37-version>3.7.3</zookeeper-37-version>
     </properties>
 
     <dependencies>

curator-test-zk38/pom.xml

@@ -31,7 +31,7 @@
     <artifactId>curator-test-zk38</artifactId>
 
     <properties>
-        <zookeeper-38-version>3.8.3</zookeeper-38-version>
+        <zookeeper-38-version>3.8.4</zookeeper-38-version>
     </properties>
 
     <dependencies>