diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt
index 10da9d7bc2..e4537a3ef7 100644
--- a/hadoop-common-project/hadoop-common/CHANGES.txt
+++ b/hadoop-common-project/hadoop-common/CHANGES.txt
@@ -748,6 +748,9 @@ Release 2.7.1 - UNRELEASED
     HADOOP-11966. Variable cygwin is undefined in hadoop-config.sh when executed
     through hadoop-daemon.sh. (cnauroth)
 
+    HADOOP-11973. Ensure ZkDelegationTokenSecretManager namespace znodes get
+    created with ACLs. (Gregory Chanan via asuresh)
+
 Release 2.7.0 - 2015-04-20
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java
index 73c3ab8239..da0e6adc62 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java
@@ -47,6 +47,7 @@
 import org.apache.curator.framework.recipes.shared.SharedCount;
 import org.apache.curator.framework.recipes.shared.VersionedValue;
 import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.utils.EnsurePath;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
@@ -298,6 +299,17 @@ public void startThreads() throws IOException {
       } catch (Exception e) {
         throw new IOException("Could not start Curator Framework", e);
       }
+    } else {
+      // If namespace parents are implicitly created, they won't have ACLs.
+      // So, let's explicitly create them.
+      CuratorFramework nullNsFw = zkClient.usingNamespace(null);
+      EnsurePath ensureNs =
+        nullNsFw.newNamespaceAwareEnsurePath("/" + zkClient.getNamespace());
+      try {
+        ensureNs.ensure(nullNsFw.getZookeeperClient());
+      } catch (Exception e) {
+        throw new IOException("Could not create namespace", e);
+      }
     }
     listenerThreadPool = Executors.newSingleThreadExecutor();
     try {
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java
index 6435c0bb4c..185a994072 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java
@@ -19,9 +19,16 @@
 package org.apache.hadoop.security.token.delegation;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.ACLProvider;
+import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.curator.test.TestingServer;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Text;
@@ -30,6 +37,10 @@
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.delegation.web.DelegationTokenIdentifier;
 import org.apache.hadoop.security.token.delegation.web.DelegationTokenManager;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -296,6 +307,55 @@ public Void call() throws Exception {
     tm1.destroy();
   }
 
+  @Test
+  public void testACLs() throws Exception {
+    DelegationTokenManager tm1;
+    String connectString = zkServer.getConnectString();
+    Configuration conf = getSecretConf(connectString);
+    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
+    String userPass = "myuser:mypass";
+    final ACL digestACL = new ACL(ZooDefs.Perms.ALL, new Id("digest",
+      DigestAuthenticationProvider.generateDigest(userPass)));
+    ACLProvider digestAclProvider = new ACLProvider() {
+      @Override
+      public List<ACL> getAclForPath(String path) { return getDefaultAcl(); }
+
+      @Override
+      public List<ACL> getDefaultAcl() {
+        List<ACL> ret = new ArrayList<ACL>();
+        ret.add(digestACL);
+        return ret;
+      }
+    };
+
+    CuratorFramework curatorFramework =
+      CuratorFrameworkFactory.builder()
+        .connectString(connectString)
+        .retryPolicy(retryPolicy)
+        .aclProvider(digestAclProvider)
+        .authorization("digest", userPass.getBytes("UTF-8"))
+        .build();
+    curatorFramework.start();
+    ZKDelegationTokenSecretManager.setCurator(curatorFramework);
+    tm1 = new DelegationTokenManager(conf, new Text("bla"));
+    tm1.init();
+
+    // check ACL
+    String workingPath = conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH);
+    verifyACL(curatorFramework, "/" + workingPath, digestACL);
+
+    tm1.destroy();
+    ZKDelegationTokenSecretManager.setCurator(null);
+    curatorFramework.close();
+  }
+
+  private void verifyACL(CuratorFramework curatorFramework,
+      String path, ACL expectedACL) throws Exception {
+    List<ACL> acls = curatorFramework.getACL().forPath(path);
+    Assert.assertEquals(1, acls.size());
+    Assert.assertEquals(expectedACL, acls.get(0));
+  }
+
   // Since it is possible that there can be a delay for the cancel token message
   // initiated by one node to reach another node.. The second node can ofcourse
   // verify with ZK directly if the token that needs verification has been