HDFS-8112. Relax permission checking for EC related operations.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirErasureCodingOp.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.server.namenode.FSDirectory.DirOp;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.security.AccessControlException;
 
 import static org.apache.hadoop.hdfs.server.common.HdfsServerConstants.XATTR_ERASURECODING_POLICY;
 
@@ -66,15 +67,15 @@ final class FSDirErasureCodingOp {
    * @return {@link HdfsFileStatus}
    * @throws IOException
    * @throws HadoopIllegalArgumentException if the policy is not enabled
+   * @throws AccessControlException if the user does not have write access
    */
   static HdfsFileStatus setErasureCodingPolicy(final FSNamesystem fsn,
       final String srcArg, final String ecPolicyName,
-      final boolean logRetryCache) throws IOException {
+      final FSPermissionChecker pc, final boolean logRetryCache)
+      throws IOException, AccessControlException {
     assert fsn.hasWriteLock();
 
     String src = srcArg;
-    FSPermissionChecker pc = null;
-    pc = fsn.getPermissionChecker();
     FSDirectory fsd = fsn.getFSDirectory();
     final INodesInPath iip;
     List<XAttr> xAttrs;
@@ -88,6 +89,10 @@ final class FSDirErasureCodingOp {
             "policies.");
       }
       iip = fsd.resolvePath(pc, src, DirOp.WRITE_LINK);
+      // Write access is required to set erasure coding policy
+      if (fsd.isPermissionEnabled()) {
+        fsd.checkPathAccess(pc, iip, FsAction.WRITE);
+      }
       src = iip.getPath();
       xAttrs = setErasureCodingPolicyXAttr(fsn, iip, ecPolicy);
     } finally {
@@ -97,7 +102,7 @@ final class FSDirErasureCodingOp {
     return fsd.getAuditFileInfo(iip);
   }
 
-  static List<XAttr> setErasureCodingPolicyXAttr(final FSNamesystem fsn,
+  private static List<XAttr> setErasureCodingPolicyXAttr(final FSNamesystem fsn,
       final INodesInPath srcIIP, ErasureCodingPolicy ecPolicy) throws IOException {
     FSDirectory fsd = fsn.getFSDirectory();
     assert fsd.hasWriteLock();
@@ -165,19 +170,24 @@ final class FSDirErasureCodingOp {
    *          cache rebuilding
    * @return {@link HdfsFileStatus}
    * @throws IOException
+   * @throws AccessControlException if the user does not have write access
    */
   static HdfsFileStatus unsetErasureCodingPolicy(final FSNamesystem fsn,
-      final String srcArg, final boolean logRetryCache) throws IOException {
+      final String srcArg, final FSPermissionChecker pc,
+      final boolean logRetryCache) throws IOException {
     assert fsn.hasWriteLock();
 
     String src = srcArg;
-    FSPermissionChecker pc = fsn.getPermissionChecker();
     FSDirectory fsd = fsn.getFSDirectory();
     final INodesInPath iip;
     List<XAttr> xAttrs;
     fsd.writeLock();
     try {
       iip = fsd.resolvePath(pc, src, DirOp.WRITE_LINK);
+      // Write access is required to unset erasure coding policy
+      if (fsd.isPermissionEnabled()) {
+        fsd.checkPathAccess(pc, iip, FsAction.WRITE);
+      }
       src = iip.getPath();
       xAttrs = removeErasureCodingPolicyXAttr(fsn, iip);
     } finally {
@@ -225,29 +235,23 @@ final class FSDirErasureCodingOp {
    * @return {@link ErasureCodingPolicy}
    * @throws IOException
    * @throws FileNotFoundException if the path does not exist.
+   * @throws AccessControlException if no read access
    */
   static ErasureCodingPolicy getErasureCodingPolicy(final FSNamesystem fsn,
-      final String src) throws IOException {
+      final String src, FSPermissionChecker pc)
+      throws IOException, AccessControlException {
     assert fsn.hasReadLock();
 
-    final INodesInPath iip = getINodesInPath(fsn, src);
+    FSDirectory fsd = fsn.getFSDirectory();
+    final INodesInPath iip = fsd.resolvePath(pc, src, DirOp.READ);
+    if (fsn.isPermissionEnabled()) {
+      fsn.getFSDirectory().checkPathAccess(pc, iip, FsAction.READ);
+    }
+
     if (iip.getLastINode() == null) {
       throw new FileNotFoundException("Path not found: " + iip.getPath());
     }
-    return getErasureCodingPolicyForPath(fsn, iip);
+    return getErasureCodingPolicyForPath(fsd, iip);
-  }
-
-  /**
-   * Check if the file or directory has an erasure coding policy.
-   *
-   * @param fsn namespace
-   * @param srcArg path
-   * @return Whether the file or directory has an erasure coding policy.
-   * @throws IOException
-   */
-  static boolean hasErasureCodingPolicy(final FSNamesystem fsn,
-      final String srcArg) throws IOException {
-    return hasErasureCodingPolicy(fsn, getINodesInPath(fsn, srcArg));
   }
 
   /**
@@ -260,22 +264,22 @@ final class FSDirErasureCodingOp {
    */
   static boolean hasErasureCodingPolicy(final FSNamesystem fsn,
       final INodesInPath iip) throws IOException {
-    return getErasureCodingPolicy(fsn, iip) != null;
+    return unprotectedGetErasureCodingPolicy(fsn, iip) != null;
   }
 
   /**
-   * Get the erasure coding policy.
+   * Get the erasure coding policy. This does not do any permission checking.
    *
    * @param fsn namespace
    * @param iip inodes in the path containing the file
    * @return {@link ErasureCodingPolicy}
    * @throws IOException
    */
-  static ErasureCodingPolicy getErasureCodingPolicy(final FSNamesystem fsn,
+  static ErasureCodingPolicy unprotectedGetErasureCodingPolicy(
-      final INodesInPath iip) throws IOException {
+      final FSNamesystem fsn, final INodesInPath iip) throws IOException {
     assert fsn.hasReadLock();
 
-    return getErasureCodingPolicyForPath(fsn, iip);
+    return getErasureCodingPolicyForPath(fsn.getFSDirectory(), iip);
   }
 
   /**
@@ -290,21 +294,9 @@ final class FSDirErasureCodingOp {
     return fsn.getErasureCodingPolicyManager().getPolicies();
   }
 
-  private static INodesInPath getINodesInPath(final FSNamesystem fsn,
+  private static ErasureCodingPolicy getErasureCodingPolicyForPath(
-      final String srcArg) throws IOException {
+      FSDirectory fsd, INodesInPath iip) throws IOException {
-    final FSDirectory fsd = fsn.getFSDirectory();
-    final FSPermissionChecker pc = fsn.getPermissionChecker();
-    INodesInPath iip = fsd.resolvePath(pc, srcArg, DirOp.READ);
-    if (fsn.isPermissionEnabled()) {
-      fsn.getFSDirectory().checkPathAccess(pc, iip, FsAction.READ);
-    }
-    return iip;
-  }
-
-  private static ErasureCodingPolicy getErasureCodingPolicyForPath(FSNamesystem fsn,
-      INodesInPath iip) throws IOException {
     Preconditions.checkNotNull(iip, "INodes cannot be null");
-    FSDirectory fsd = fsn.getFSDirectory();
     fsd.readLock();
     try {
       List<INode> inodes = iip.getReadOnlyINodes();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java

@@ -172,7 +172,7 @@ class FSDirStatAndListingOp {
       final FileEncryptionInfo feInfo =
           FSDirEncryptionZoneOp.getFileEncryptionInfo(fsd, iip);
       final ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.
-          getErasureCodingPolicy(fsd.getFSNamesystem(), iip);
+          unprotectedGetErasureCodingPolicy(fsd.getFSNamesystem(), iip);
 
       final LocatedBlocks blocks = bm.createLocatedBlocks(
           inode.getBlocks(iip.getPathSnapshotId()), fileSize, isUc, offset,
@@ -413,7 +413,7 @@ class FSDirStatAndListingOp {
     FileEncryptionInfo feInfo = null;
 
     final ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp
-        .getErasureCodingPolicy(fsd.getFSNamesystem(), iip);
+        .unprotectedGetErasureCodingPolicy(fsd.getFSNamesystem(), iip);
 
     if (node.isFile()) {
       final INodeFile fileNode = node.asFile();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java

@@ -190,7 +190,8 @@ class FSDirWriteFileOp {
     blockType = pendingFile.getBlockType();
     ErasureCodingPolicy ecPolicy = null;
     if (blockType == BlockType.STRIPED) {
-      ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(fsn, src);
+      ecPolicy =
+          FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(fsn, iip);
       numTargets = (short) (ecPolicy.getSchema().getNumDataUnits()
           + ecPolicy.getSchema().getNumParityUnits());
     } else {
@@ -418,7 +419,7 @@ class FSDirWriteFileOp {
       // check if the file has an EC policy
       boolean isStriped = false;
       ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.
-          getErasureCodingPolicy(fsd.getFSNamesystem(), existing);
+          unprotectedGetErasureCodingPolicy(fsd.getFSNamesystem(), existing);
       if (ecPolicy != null) {
         isStriped = true;
       }
@@ -475,7 +476,8 @@ class FSDirWriteFileOp {
       // associate new last block for the file
       final BlockInfo blockInfo;
       if (blockType == BlockType.STRIPED) {
-        ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(
+        ErasureCodingPolicy ecPolicy =
+            FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(
                 fsd.getFSNamesystem(), inodesInPath);
         short numDataUnits = (short) ecPolicy.getNumDataUnits();
         short numParityUnits = (short) ecPolicy.getNumParityUnits();
@@ -529,7 +531,7 @@ class FSDirWriteFileOp {
     try {
       boolean isStriped = false;
       ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.
-          getErasureCodingPolicy(fsd.getFSNamesystem(), existing);
+          unprotectedGetErasureCodingPolicy(fsd.getFSNamesystem(), existing);
       if (ecPolicy != null) {
         isStriped = true;
       }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java

@@ -957,7 +957,8 @@ public class FSDirectory implements Closeable {
     final short replicationFactor;
     if (fileINode.isStriped()) {
       final ErasureCodingPolicy ecPolicy =
-          FSDirErasureCodingOp.getErasureCodingPolicy(namesystem, iip);
+          FSDirErasureCodingOp
+              .unprotectedGetErasureCodingPolicy(namesystem, iip);
       final short numDataUnits = (short) ecPolicy.getNumDataUnits();
       final short numParityUnits = (short) ecPolicy.getNumParityUnits();
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java

@@ -415,7 +415,8 @@ public class FSEditLogLoader {
       // Update the salient file attributes.
       newFile.setAccessTime(addCloseOp.atime, Snapshot.CURRENT_STATE_ID);
       newFile.setModificationTime(addCloseOp.mtime, Snapshot.CURRENT_STATE_ID);
-      ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(
+      ErasureCodingPolicy ecPolicy =
+          FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(
               fsDir.getFSNamesystem(), iip);
       updateBlocks(fsDir, addCloseOp, iip, newFile, ecPolicy);
       break;
@@ -437,7 +438,8 @@ public class FSEditLogLoader {
       // Update the salient file attributes.
       file.setAccessTime(addCloseOp.atime, Snapshot.CURRENT_STATE_ID);
       file.setModificationTime(addCloseOp.mtime, Snapshot.CURRENT_STATE_ID);
-      ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(
+      ErasureCodingPolicy ecPolicy =
+          FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(
               fsDir.getFSNamesystem(), iip);
       updateBlocks(fsDir, addCloseOp, iip, file, ecPolicy);
 
@@ -496,7 +498,8 @@ public class FSEditLogLoader {
       INodesInPath iip = fsDir.getINodesInPath(path, DirOp.READ);
       INodeFile oldFile = INodeFile.valueOf(iip.getLastINode(), path);
       // Update in-memory data structures
-      ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(
+      ErasureCodingPolicy ecPolicy =
+          FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(
               fsDir.getFSNamesystem(), iip);
       updateBlocks(fsDir, updateOp, iip, oldFile, ecPolicy);
 
@@ -515,7 +518,8 @@ public class FSEditLogLoader {
       INodesInPath iip = fsDir.getINodesInPath(path, DirOp.READ);
       INodeFile oldFile = INodeFile.valueOf(iip.getLastINode(), path);
       // add the new block to the INodeFile
-      ErasureCodingPolicy ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(
+      ErasureCodingPolicy ecPolicy =
+          FSDirErasureCodingOp.unprotectedGetErasureCodingPolicy(
               fsDir.getFSNamesystem(), iip);
       addNewBlock(addBlockOp, oldFile, ecPolicy);
       break;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -6782,16 +6782,16 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       final boolean logRetryCache) throws IOException,
       UnresolvedLinkException, SafeModeException, AccessControlException {
     final String operationName = "setErasureCodingPolicy";
-    checkSuperuserPrivilege();
     checkOperation(OperationCategory.WRITE);
     HdfsFileStatus resultingStat = null;
+    final FSPermissionChecker pc = getPermissionChecker();
     boolean success = false;
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot set erasure coding policy on " + srcArg);
       resultingStat = FSDirErasureCodingOp.setErasureCodingPolicy(this,
-          srcArg, ecPolicyName, logRetryCache);
+          srcArg, ecPolicyName, pc, logRetryCache);
       success = true;
     } catch (AccessControlException ace) {
       logAuditEvent(success, operationName, srcArg, null,
@@ -6818,16 +6818,16 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       final boolean logRetryCache) throws IOException,
       UnresolvedLinkException, SafeModeException, AccessControlException {
     final String operationName = "unsetErasureCodingPolicy";
-    checkSuperuserPrivilege();
     checkOperation(OperationCategory.WRITE);
     HdfsFileStatus resultingStat = null;
+    final FSPermissionChecker pc = getPermissionChecker();
     boolean success = false;
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot unset erasure coding policy on " + srcArg);
       resultingStat = FSDirErasureCodingOp.unsetErasureCodingPolicy(this,
-          srcArg, logRetryCache);
+          srcArg, pc, logRetryCache);
       success = true;
     } catch (AccessControlException ace) {
       logAuditEvent(success, operationName, srcArg, null,
@@ -6849,10 +6849,11 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
   ErasureCodingPolicy getErasureCodingPolicy(String src)
       throws AccessControlException, UnresolvedLinkException, IOException {
     checkOperation(OperationCategory.READ);
+    FSPermissionChecker pc = getPermissionChecker();
     readLock();
     try {
       checkOperation(OperationCategory.READ);
-      return FSDirErasureCodingOp.getErasureCodingPolicy(this, src);
+      return FSDirErasureCodingOp.getErasureCodingPolicy(this, src, pc);
     } finally {
       readUnlock("getErasureCodingPolicy");
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestErasureCodingPolicies.java

@@ -19,6 +19,7 @@ package org.apache.hadoop.hdfs;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.protocol.DirectoryListing;
@@ -30,12 +31,17 @@ import org.apache.hadoop.hdfs.client.HdfsAdmin;
 import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 import org.apache.hadoop.hdfs.server.namenode.INodeFile;
 import org.apache.hadoop.io.erasurecode.ECSchema;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.Timeout;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
 import java.util.Arrays;
 import java.util.Collection;
 
@@ -51,6 +57,9 @@ public class TestErasureCodingPolicies {
       ErasureCodingPolicyManager.getSystemDefaultPolicy();
   private FSNamesystem namesystem;
 
+  @Rule
+  public Timeout timeout = new Timeout(60 * 1000);
+
   @Before
   public void setupCluster() throws IOException {
     conf = new HdfsConfiguration();
@@ -74,7 +83,7 @@ public class TestErasureCodingPolicies {
    * for pre-existing files (with replicated blocks) in an EC dir, getListing
    * should report them as non-ec.
    */
-  @Test(timeout=60000)
+  @Test
   public void testReplicatedFileUnderECDir() throws IOException {
     final Path dir = new Path("/ec");
     final Path replicatedFile = new Path(dir, "replicatedFile");
@@ -128,7 +137,7 @@ public class TestErasureCodingPolicies {
     assertNotNull(files[1].getErasureCodingPolicy());
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testBasicSetECPolicy()
       throws IOException, InterruptedException {
     final Path testDir = new Path("/ec");
@@ -182,7 +191,7 @@ public class TestErasureCodingPolicies {
     }
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testMoveValidity() throws IOException, InterruptedException {
     final Path srcECDir = new Path("/srcEC");
     final Path dstECDir = new Path("/dstEC");
@@ -219,7 +228,7 @@ public class TestErasureCodingPolicies {
     fs.rename(nonECFile, dstECDir);
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testReplication() throws IOException {
     final Path testDir = new Path("/ec");
     fs.mkdir(testDir, FsPermission.getDirDefault());
@@ -237,7 +246,7 @@ public class TestErasureCodingPolicies {
     assertEquals(policy, fs.getErasureCodingPolicy(fooFile));
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testGetErasureCodingPolicyWithSystemDefaultECPolicy() throws Exception {
     String src = "/ec";
     final Path ecDir = new Path(src);
@@ -254,7 +263,7 @@ public class TestErasureCodingPolicies {
     verifyErasureCodingInfo(src + "/child1", sysDefaultECPolicy);
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testGetErasureCodingPolicy() throws Exception {
     ErasureCodingPolicy[] sysECPolicies =
         ErasureCodingPolicyManager.getSystemPolicies();
@@ -284,13 +293,13 @@ public class TestErasureCodingPolicies {
         usingECPolicy, ecPolicy);
   }
 
-  @Test(timeout = 60000)
+  @Test
-  public void testCreationErasureCodingZoneWithInvalidPolicy()
+  public void testSetInvalidPolicy()
       throws IOException {
     ECSchema rsSchema = new ECSchema("rs", 4, 2);
     String policyName = "RS-4-2-128k";
     int cellSize = 128 * 1024;
-    ErasureCodingPolicy ecPolicy=
+    ErasureCodingPolicy ecPolicy =
         new ErasureCodingPolicy(policyName, rsSchema, cellSize, (byte) -1);
     String src = "/ecDir4-2";
     final Path ecDir = new Path(src);
@@ -305,7 +314,7 @@ public class TestErasureCodingPolicies {
     }
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testGetAllErasureCodingPolicies() throws Exception {
     ErasureCodingPolicy[] sysECPolicies = ErasureCodingPolicyManager
         .getSystemPolicies();
@@ -315,7 +324,7 @@ public class TestErasureCodingPolicies {
         allECPolicies.containsAll(Arrays.asList(sysECPolicies)));
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testGetErasureCodingPolicyOnANonExistentFile() throws Exception {
     Path path = new Path("/ecDir");
     try {
@@ -335,7 +344,7 @@ public class TestErasureCodingPolicies {
     }
   }
 
-  @Test(timeout = 60000)
+  @Test
   public void testMultiplePoliciesCoExist() throws Exception {
     ErasureCodingPolicy[] sysPolicies =
         ErasureCodingPolicyManager.getSystemPolicies();
@@ -355,4 +364,124 @@ public class TestErasureCodingPolicies {
       }
     }
   }
+
+  @Test
+  public void testPermissions() throws Exception {
+    UserGroupInformation user =
+        UserGroupInformation.createUserForTesting("ecuser",
+            new String[]{"ecgroup"});
+    FileSystem userfs = user.doAs(new PrivilegedExceptionAction<FileSystem>() {
+      @Override
+      public FileSystem run() throws Exception {
+        return FileSystem.get(conf);
+      }
+    });
+    HdfsAdmin useradmin = user.doAs(new PrivilegedExceptionAction<HdfsAdmin>() {
+      @Override
+      public HdfsAdmin run() throws Exception {
+        return new HdfsAdmin(userfs.getUri(), conf);
+      }
+    });
+
+    // Create dir and set an EC policy, create an EC file
+    Path ecdir = new Path("/ecdir");
+    Path ecfile = new Path(ecdir, "ecfile");
+    fs.setPermission(new Path("/"), new FsPermission((short)0777));
+    userfs.mkdirs(ecdir);
+    final String ecPolicyName =
+        ErasureCodingPolicyManager.getSystemPolicies()[0].getName();
+    useradmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+    assertEquals("Policy not present on dir",
+        ecPolicyName,
+        useradmin.getErasureCodingPolicy(ecdir).getName());
+    userfs.create(ecfile).close();
+    assertEquals("Policy not present on file",
+        ecPolicyName,
+        useradmin.getErasureCodingPolicy(ecfile).getName());
+
+    // Unset and re-set
+    useradmin.unsetErasureCodingPolicy(ecdir);
+    useradmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+
+    // Change write permissions and make sure set and unset are denied
+    userfs.setPermission(ecdir, new FsPermission((short)0555));
+    try {
+      useradmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+      fail("Should not be able to setECPolicy without write permissions");
+    } catch (AccessControlException e) {
+      // pass
+    }
+    try {
+      useradmin.unsetErasureCodingPolicy(ecdir);
+      fail("Should not be able to unsetECPolicy without write permissions");
+    } catch (AccessControlException e) {
+      // pass
+    }
+
+    // Change the permissions again, check that set and unset work
+    userfs.setPermission(ecdir, new FsPermission((short)0640));
+    useradmin.unsetErasureCodingPolicy(ecdir);
+    useradmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+
+    // Set, unset, and get with another user should be unauthorized
+    UserGroupInformation nobody =
+        UserGroupInformation.createUserForTesting("nobody",
+            new String[]{"nogroup"});
+    HdfsAdmin noadmin = nobody.doAs(new PrivilegedExceptionAction<HdfsAdmin>() {
+      @Override
+      public HdfsAdmin run() throws Exception {
+        return new HdfsAdmin(userfs.getUri(), conf);
+      }
+    });
+    try {
+      noadmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+      fail("Should not be able to setECPolicy without write permissions");
+    } catch (AccessControlException e) {
+      // pass
+    }
+    try {
+      noadmin.unsetErasureCodingPolicy(ecdir);
+      fail("Should not be able to unsetECPolicy without write permissions");
+    } catch (AccessControlException e) {
+      // pass
+    }
+    try {
+      noadmin.getErasureCodingPolicy(ecdir);
+      fail("Should not be able to getECPolicy without write permissions");
+    } catch (AccessControlException e) {
+      // pass
+    }
+
+    // superuser can do whatever it wants
+    userfs.setPermission(ecdir, new FsPermission((short)0000));
+    HdfsAdmin superadmin = new HdfsAdmin(fs.getUri(), conf);
+    superadmin.unsetErasureCodingPolicy(ecdir);
+    superadmin.setErasureCodingPolicy(ecdir, ecPolicyName);
+    superadmin.getErasureCodingPolicy(ecdir);
+
+    // Normal user no longer has access
+    try {
+      useradmin.getErasureCodingPolicy(ecdir);
+      fail("Normal user should not have access");
+    } catch (AccessControlException e) {
+      // pass
+    }
+    try {
+      useradmin.setErasureCodingPolicy(ecfile, ecPolicyName);
+      fail("Normal user should not have access");
+    } catch (AccessControlException e) {
+      // pass
+    }
+    try {
+      useradmin.unsetErasureCodingPolicy(ecfile);
+      fail("Normal user should not have access");
+    } catch (AccessControlException e) {
+      // pass
+    }
+
+    // Everyone has access to getting the list of EC policies
+    useradmin.getErasureCodingPolicies();
+    noadmin.getErasureCodingPolicies();
+    superadmin.getErasureCodingPolicies();
+  }
 }