From eb791700eedadbac7d8baff5648156bafa08fc01 Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Sat, 2 Apr 2011 14:41:10 +0000
Subject: [PATCH] HBASE-3238 HBase needs to have the CREATE permission on the
 parent of its ZooKeeper parent znode

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1088038 13f79535-47bb-0310-9956-ffa450edef68
---
 CHANGES.txt                                   |  2 ++
 .../apache/hadoop/hbase/zookeeper/ZKUtil.java | 10 ++++++
 .../apache/hadoop/hbase/TestZooKeeper.java    | 34 +++++++++++++++++++
 3 files changed, 46 insertions(+)

diff --git a/CHANGES.txt b/CHANGES.txt
index 708a926bab7..456f5c0eb74 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -58,6 +58,8 @@ Release 0.91.0 - Unreleased
                (Ted Yu via Stack)
    HBASE-3712  HTable.close() doesn't shutdown thread pool
                (Ted Yu via Stack)
+   HBASE-3238  HBase needs to have the CREATE permission on the parent of its
+               ZooKeeper parent znode (Alex Newman via Stack)
 
   IMPROVEMENTS
    HBASE-3290  Max Compaction Size (Nicolas Spiegelberg via Stack)  
diff --git a/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java b/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java
index 08748f81027..c01fe679412 100644
--- a/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java
+++ b/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java
@@ -902,6 +902,16 @@ public class ZKUtil {
       zkw.getZooKeeper().create(znode, new byte[0], Ids.OPEN_ACL_UNSAFE,
           CreateMode.PERSISTENT);
     } catch(KeeperException.NodeExistsException nee) {
+    } catch(KeeperException.NoAuthException nee){
+      try {
+        if (null == zkw.getZooKeeper().exists(znode, false)) {
+          // If we failed to create the file and it does not already exist.
+          throw(nee);
+        }
+      } catch (InterruptedException ie) {
+        zkw.interruptedException(ie);
+      }
+
     } catch(InterruptedException ie) {
       zkw.interruptedException(ie);
     }
diff --git a/src/test/java/org/apache/hadoop/hbase/TestZooKeeper.java b/src/test/java/org/apache/hadoop/hbase/TestZooKeeper.java
index 5852ab7570a..7b19356741a 100644
--- a/src/test/java/org/apache/hadoop/hbase/TestZooKeeper.java
+++ b/src/test/java/org/apache/hadoop/hbase/TestZooKeeper.java
@@ -40,7 +40,9 @@ import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.ZooKeeper.States;
 import org.junit.AfterClass;
@@ -234,4 +236,36 @@ public class TestZooKeeper {
     String reconstructedKey = ZKUtil.getZooKeeperClusterKey(conf);
     assertEquals(key, reconstructedKey);
   }
+
+  /**
+   * A test for HBASE-3238
+   * @throws IOException A connection attempt to zk failed
+   * @throws InterruptedException One of the non ZKUtil actions was interrupted
+   * @throws KeeperException Any of the zookeeper connections had a
+   * KeeperException
+   */
+  @Test
+  public void testCreateSilentIsReallySilent() throws InterruptedException,
+      KeeperException, IOException {
+    Configuration c = TEST_UTIL.getConfiguration();
+
+    String aclZnode = "/aclRoot";
+    String quorumServers = ZKConfig.getZKQuorumServersString(c);
+    int sessionTimeout = 5 * 1000; // 5 seconds
+    ZooKeeper zk = new ZooKeeper(quorumServers, sessionTimeout, EmptyWatcher.instance);
+    zk.addAuthInfo("digest", "hbase:rox".getBytes());
+
+    // Assumes the  root of the ZooKeeper space is writable as it creates a node
+    // wherever the cluster home is defined.
+    ZooKeeperWatcher zk2 = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(),
+        "testMasterAddressManagerFromZK",
+        null);
+
+    // I set this acl after the attempted creation of the cluster home node.
+    zk.setACL("/", ZooDefs.Ids.CREATOR_ALL_ACL, -1);
+    zk.create(aclZnode, null, ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+    zk.close();
+
+    ZKUtil.createAndFailSilent(zk2, aclZnode);
+ }
 }
\ No newline at end of file