HDFS-5608. WebHDFS: implement ACL APIs. Contributed by Sachin Jose and Renil Joseph.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-4685@1563013 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES-HDFS-4685.txt

@@ -47,6 +47,9 @@ HDFS-4685 (Unreleased)
     HDFS-5702. FsShell Cli: Add XML based End-to-End test for getfacl and
     setfacl commands. (Vinay via cnauroth)
 
+    HDFS-5608. WebHDFS: implement ACL APIs.
+    (Sachin Jose and Renil Joseph via cnauroth)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -69,6 +69,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int     DFS_CLIENT_RETRY_TIMES_GET_LAST_BLOCK_LENGTH_DEFAULT = 3;
   public static final String  DFS_CLIENT_RETRY_INTERVAL_GET_LAST_BLOCK_LENGTH = "dfs.client.retry.interval-ms.get-last-block-length";
   public static final int     DFS_CLIENT_RETRY_INTERVAL_GET_LAST_BLOCK_LENGTH_DEFAULT = 4000;
+  public static final String DFS_WEBHDFS_ACL_PERMISSION_PATTERN_DEFAULT =
+      "^(default:)?(user|group|mask|other):[[A-Za-z_][A-Za-z0-9._-]]*:([rwx-]{3})?(,(default:)?(user|group|mask|other):[[A-Za-z_][A-Za-z0-9._-]]*:([rwx-]{3})?)*$";
 
   // HA related configuration
   public static final String  DFS_CLIENT_FAILOVER_PROXY_PROVIDER_KEY_PREFIX = "dfs.client.failover.proxy.provider";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/web/resources/NamenodeWebHdfsMethods.java

@@ -53,6 +53,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.hdfs.StorageType;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.DirectoryListing;
@@ -71,6 +72,7 @@ import org.apache.hadoop.hdfs.web.ParamFilter;
 import org.apache.hadoop.hdfs.web.SWebHdfsFileSystem;
 import org.apache.hadoop.hdfs.web.WebHdfsFileSystem;
 import org.apache.hadoop.hdfs.web.resources.AccessTimeParam;
+import org.apache.hadoop.hdfs.web.resources.AclPermissionParam;
 import org.apache.hadoop.hdfs.web.resources.BlockSizeParam;
 import org.apache.hadoop.hdfs.web.resources.BufferSizeParam;
 import org.apache.hadoop.hdfs.web.resources.ConcatSourcesParam;
@@ -301,12 +303,14 @@ public class NamenodeWebHdfsMethods {
       @QueryParam(CreateParentParam.NAME) @DefaultValue(CreateParentParam.DEFAULT)
           final CreateParentParam createParent,
       @QueryParam(TokenArgumentParam.NAME) @DefaultValue(TokenArgumentParam.DEFAULT)
-          final TokenArgumentParam delegationTokenArgument
-      ) throws IOException, InterruptedException {
+          final TokenArgumentParam delegationTokenArgument,
+      @QueryParam(AclPermissionParam.NAME) @DefaultValue(AclPermissionParam.DEFAULT) 
+          final AclPermissionParam aclPermission
+          )throws IOException, InterruptedException {
     return put(ugi, delegation, username, doAsUser, ROOT, op, destination,
         owner, group, permission, overwrite, bufferSize, replication,
         blockSize, modificationTime, accessTime, renameOptions, createParent,
-        delegationTokenArgument);
+        delegationTokenArgument,aclPermission);
   }
 
   /** Handle HTTP PUT request. */
@@ -350,12 +354,14 @@ public class NamenodeWebHdfsMethods {
       @QueryParam(CreateParentParam.NAME) @DefaultValue(CreateParentParam.DEFAULT)
           final CreateParentParam createParent,
       @QueryParam(TokenArgumentParam.NAME) @DefaultValue(TokenArgumentParam.DEFAULT)
-          final TokenArgumentParam delegationTokenArgument
+          final TokenArgumentParam delegationTokenArgument,
+      @QueryParam(AclPermissionParam.NAME) @DefaultValue(AclPermissionParam.DEFAULT) 
+          final AclPermissionParam aclPermission
       ) throws IOException, InterruptedException {
 
     init(ugi, delegation, username, doAsUser, path, op, destination, owner,
         group, permission, overwrite, bufferSize, replication, blockSize,
-        modificationTime, accessTime, renameOptions, delegationTokenArgument);
+        modificationTime, accessTime, renameOptions, delegationTokenArgument,aclPermission);
 
     return ugi.doAs(new PrivilegedExceptionAction<Response>() {
       @Override
@@ -366,7 +372,7 @@ public class NamenodeWebHdfsMethods {
               path.getAbsolutePath(), op, destination, owner, group,
               permission, overwrite, bufferSize, replication, blockSize,
               modificationTime, accessTime, renameOptions, createParent,
-              delegationTokenArgument);
+              delegationTokenArgument,aclPermission);
         } finally {
           REMOTE_ADDRESS.set(null);
         }
@@ -393,7 +399,8 @@ public class NamenodeWebHdfsMethods {
       final AccessTimeParam accessTime,
       final RenameOptionSetParam renameOptions,
       final CreateParentParam createParent,
-      final TokenArgumentParam delegationTokenArgument
+      final TokenArgumentParam delegationTokenArgument,
+      final AclPermissionParam aclPermission
       ) throws IOException, URISyntaxException {
 
     final Configuration conf = (Configuration)context.getAttribute(JspHelper.CURRENT_CONF);
@@ -473,6 +480,26 @@ public class NamenodeWebHdfsMethods {
       np.cancelDelegationToken(token);
       return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
     }
+    case MODIFYACLENTRIES: {
+      np.modifyAclEntries(fullpath, aclPermission.getAclPermission(true));
+      return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
+    }
+    case REMOVEACLENTRIES: {
+      np.removeAclEntries(fullpath, aclPermission.getAclPermission(false));
+      return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
+    }
+    case REMOVEDEFAULTACL: {
+      np.removeDefaultAcl(fullpath);
+      return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
+    }
+    case REMOVEACL: {
+      np.removeAcl(fullpath);
+      return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
+    }
+    case SETACL: {
+      np.setAcl(fullpath, aclPermission.getAclPermission(true));
+      return Response.ok().type(MediaType.APPLICATION_OCTET_STREAM).build();
+    }
     default:
       throw new UnsupportedOperationException(op + " is not supported");
     }
@@ -713,6 +740,15 @@ public class NamenodeWebHdfsMethods {
           WebHdfsFileSystem.getHomeDirectoryString(ugi));
       return Response.ok(js).type(MediaType.APPLICATION_JSON).build();
     }
+    case GETACLSTATUS: {
+      AclStatus status = np.getAclStatus(fullpath);
+      if (status == null) {
+        throw new FileNotFoundException("File does not exist: " + fullpath);
+      }
+
+      final String js = JsonUtil.toJsonString(status);
+      return Response.ok(js).type(MediaType.APPLICATION_JSON).build();
+    }
     default:
       throw new UnsupportedOperationException(op + " is not supported");
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/JsonUtil.java

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hdfs.web;
 
 import org.apache.hadoop.fs.*;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.protocol.*;
@@ -613,4 +615,44 @@ public class JsonUtil {
 
     return checksum;
   }
+  /** Convert a AclStatus object to a Json string. */
+  public static String toJsonString(final AclStatus status) {
+    if (status == null) {
+      return null;
+    }
+
+    final Map<String, Object> m = new TreeMap<String, Object>();
+    m.put("owner", status.getOwner());
+    m.put("group", status.getGroup());
+    m.put("stickyBit", status.isStickyBit());
+    m.put("entries", status.getEntries());
+    final Map<String, Map<String, Object>> finalMap =
+        new TreeMap<String, Map<String, Object>>();
+    finalMap.put(AclStatus.class.getSimpleName(), m);
+    return JSON.toString(finalMap);
+  }
+
+  /** Convert a Json map to a AclStatus object. */
+  public static AclStatus toAclStatus(final Map<?, ?> json) {
+    if (json == null) {
+      return null;
+    }
+
+    final Map<?, ?> m = (Map<?, ?>) json.get(AclStatus.class.getSimpleName());
+
+    AclStatus.Builder aclStatusBuilder = new AclStatus.Builder();
+    aclStatusBuilder.owner((String) m.get("owner"));
+    aclStatusBuilder.group((String) m.get("group"));
+    aclStatusBuilder.stickyBit((Boolean) m.get("stickyBit"));
+
+    final Object[] entries = (Object[]) m.get("entries");
+
+    List<AclEntry> aclEntryList = new ArrayList<AclEntry>();
+    for (int i = 0; i < entries.length; i++) {
+      AclEntry aclEntry = AclEntry.parseAclEntry((String) entries[i], true);
+      aclEntryList.add(aclEntry);
+    }
+    aclStatusBuilder.addEntries(aclEntryList);
+    return aclStatusBuilder.build();
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java

@@ -49,6 +49,8 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.MD5MD5CRC32FileChecksum;
 import org.apache.hadoop.fs.Options;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSUtil;
@@ -57,6 +59,7 @@ import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.server.namenode.SafeModeException;
 import org.apache.hadoop.hdfs.web.resources.AccessTimeParam;
+import org.apache.hadoop.hdfs.web.resources.AclPermissionParam;
 import org.apache.hadoop.hdfs.web.resources.BlockSizeParam;
 import org.apache.hadoop.hdfs.web.resources.BufferSizeParam;
 import org.apache.hadoop.hdfs.web.resources.ConcatSourcesParam;
@@ -694,6 +697,17 @@ public class WebHdfsFileSystem extends FileSystem
         f.getFullPath(parent).makeQualified(getUri(), getWorkingDirectory()));
   }
 
+  @Override
+  public AclStatus getAclStatus(Path f) throws IOException {
+    final HttpOpParam.Op op = GetOpParam.Op.GETACLSTATUS;
+    final Map<?, ?> json = run(op, f);
+    AclStatus status = JsonUtil.toAclStatus(json);
+    if (status == null) {
+      throw new FileNotFoundException("File does not exist: " + f);
+    }
+    return status;
+  }
+
   @Override
   public boolean mkdirs(Path f, FsPermission permission) throws IOException {
     statistics.incrementWriteOps(1);
@@ -754,6 +768,44 @@ public class WebHdfsFileSystem extends FileSystem
     run(op, p, new PermissionParam(permission));
   }
 
+  @Override
+  public void modifyAclEntries(Path path, List<AclEntry> aclSpec)
+      throws IOException {
+    statistics.incrementWriteOps(1);
+    final HttpOpParam.Op op = PutOpParam.Op.MODIFYACLENTRIES;
+    run(op, path, new AclPermissionParam(aclSpec));
+  }
+
+  @Override
+  public void removeAclEntries(Path path, List<AclEntry> aclSpec)
+      throws IOException {
+    statistics.incrementWriteOps(1);
+    final HttpOpParam.Op op = PutOpParam.Op.REMOVEACLENTRIES;
+    run(op, path, new AclPermissionParam(aclSpec));
+  }
+
+  @Override
+  public void removeDefaultAcl(Path path) throws IOException {
+    statistics.incrementWriteOps(1);
+    final HttpOpParam.Op op = PutOpParam.Op.REMOVEDEFAULTACL;
+    run(op, path);
+  }
+
+  @Override
+  public void removeAcl(Path path) throws IOException {
+    statistics.incrementWriteOps(1);
+    final HttpOpParam.Op op = PutOpParam.Op.REMOVEACL;
+    run(op, path);
+  }
+
+  @Override
+  public void setAcl(final Path p, final List<AclEntry> aclSpec)
+      throws IOException {
+    statistics.incrementWriteOps(1);
+    final HttpOpParam.Op op = PutOpParam.Op.SETACL;
+    run(op, p, new AclPermissionParam(aclSpec));
+  }
+
   @Override
   public boolean setReplication(final Path p, final short replication
      ) throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/resources/AclPermissionParam.java

@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.web.resources;
+
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_WEBHDFS_ACL_PERMISSION_PATTERN_DEFAULT;
+
+import java.util.List;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.commons.lang.StringUtils;
+
+/** AclPermission parameter. */
+public class AclPermissionParam extends StringParam {
+  /** Parameter name. */
+  public static final String NAME = "aclspec";
+  /** Default parameter value. */
+  public static final String DEFAULT = "";
+
+  private static Domain DOMAIN = new Domain(NAME,
+      Pattern.compile(DFS_WEBHDFS_ACL_PERMISSION_PATTERN_DEFAULT));
+
+  /**
+   * Constructor.
+   * 
+   * @param str a string representation of the parameter value.
+   */
+  public AclPermissionParam(final String str) {
+    super(DOMAIN, str == null || str.equals(DEFAULT) ? null : str);
+  }
+
+  public AclPermissionParam(List<AclEntry> acl) {
+    super(DOMAIN,parseAclSpec(acl).equals(DEFAULT) ? null : parseAclSpec(acl));
+  }
+
+  @Override
+  public String getName() {
+    return NAME;
+  }
+
+  public List<AclEntry> getAclPermission(boolean includePermission) {
+    final String v = getValue();
+    return (v != null ? AclEntry.parseAclSpec(v, includePermission) : AclEntry
+        .parseAclSpec(DEFAULT, includePermission));
+  }
+
+  /**
+   * Parse the list of AclEntry and returns aclspec.
+   * 
+   * @param List <AclEntry>
+   * @return String
+   */
+  private static String parseAclSpec(List<AclEntry> aclEntry) {
+    return StringUtils.join(aclEntry, ",");
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/resources/GetOpParam.java

@@ -35,6 +35,7 @@ public class GetOpParam extends HttpOpParam<GetOpParam.Op> {
 
     /** GET_BLOCK_LOCATIONS is a private unstable op. */
     GET_BLOCK_LOCATIONS(false, HttpURLConnection.HTTP_OK),
+    GETACLSTATUS(false, HttpURLConnection.HTTP_OK),
 
     NULL(false, HttpURLConnection.HTTP_NOT_IMPLEMENTED);
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/resources/PutOpParam.java

@@ -37,6 +37,12 @@ public class PutOpParam extends HttpOpParam<PutOpParam.Op> {
     RENEWDELEGATIONTOKEN(false, HttpURLConnection.HTTP_OK, true),
     CANCELDELEGATIONTOKEN(false, HttpURLConnection.HTTP_OK, true),
     
+    MODIFYACLENTRIES(false, HttpURLConnection.HTTP_OK),
+    REMOVEACLENTRIES(false, HttpURLConnection.HTTP_OK),
+    REMOVEDEFAULTACL(false, HttpURLConnection.HTTP_OK),
+    REMOVEACL(false, HttpURLConnection.HTTP_OK),
+    SETACL(false, HttpURLConnection.HTTP_OK),
+    
     NULL(false, HttpURLConnection.HTTP_NOT_IMPLEMENTED);
 
     final boolean doOutputAndRedirect;

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestJsonUtil.java

@@ -17,11 +17,19 @@
  */
 package org.apache.hadoop.hdfs.web;
 
+import static org.apache.hadoop.fs.permission.AclEntryScope.*;
+import static org.apache.hadoop.fs.permission.AclEntryType.*;
+import static org.apache.hadoop.fs.permission.FsAction.*;
+import static org.apache.hadoop.hdfs.server.namenode.AclTestHelpers.*;
+
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
@@ -32,6 +40,8 @@ import org.junit.Assert;
 import org.junit.Test;
 import org.mortbay.util.ajax.JSON;
 
+import com.google.common.collect.Lists;
+
 public class TestJsonUtil {
   static FileStatus toFileStatus(HdfsFileStatus f, String parent) {
     return new FileStatus(f.getLen(), f.isDir(), f.getReplication(),
@@ -135,6 +145,47 @@ public class TestJsonUtil {
     response.put("ipAddr", "127.0.0.1");
     checkDecodeFailure(response);
   }
+  
+  @Test
+  public void testToAclStatus() {
+    String jsonString =
+        "{\"AclStatus\":{\"entries\":[\"user::rwx\",\"user:user1:rw-\",\"group::rw-\",\"other::r-x\"],\"group\":\"supergroup\",\"owner\":\"testuser\",\"stickyBit\":false}}";
+    Map<?, ?> json = (Map<?, ?>) JSON.parse(jsonString);
+
+    List<AclEntry> aclSpec =
+        Lists.newArrayList(aclEntry(ACCESS, USER, ALL),
+            aclEntry(ACCESS, USER, "user1", READ_WRITE),
+            aclEntry(ACCESS, GROUP, READ_WRITE),
+            aclEntry(ACCESS, OTHER, READ_EXECUTE));
+
+    AclStatus.Builder aclStatusBuilder = new AclStatus.Builder();
+    aclStatusBuilder.owner("testuser");
+    aclStatusBuilder.group("supergroup");
+    aclStatusBuilder.addEntries(aclSpec);
+    aclStatusBuilder.stickyBit(false);
+
+    Assert.assertEquals("Should be equal", aclStatusBuilder.build(),
+        JsonUtil.toAclStatus(json));
+  }
+
+  @Test
+  public void testToJsonFromAclStatus() {
+    String jsonString =
+        "{\"AclStatus\":{\"entries\":[\"user:user1:rwx\",\"group::rw-\"],\"group\":\"supergroup\",\"owner\":\"testuser\",\"stickyBit\":false}}";
+    AclStatus.Builder aclStatusBuilder = new AclStatus.Builder();
+    aclStatusBuilder.owner("testuser");
+    aclStatusBuilder.group("supergroup");
+    aclStatusBuilder.stickyBit(false);
+
+    List<AclEntry> aclSpec =
+        Lists.newArrayList(aclEntry(ACCESS, USER,"user1", ALL),
+            aclEntry(ACCESS, GROUP, READ_WRITE));
+
+    aclStatusBuilder.addEntries(aclSpec);
+    Assert.assertEquals(jsonString,
+        JsonUtil.toJsonString(aclStatusBuilder.build()));
+
+  }
 
   private void checkDecodeFailure(Map<String, Object> map) {
     try {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestWebHDFS.java

@@ -18,9 +18,25 @@
 
 package org.apache.hadoop.hdfs.web;
 
+import static org.apache.hadoop.fs.permission.AclEntryScope.ACCESS;
+import static org.apache.hadoop.fs.permission.AclEntryScope.DEFAULT;
+import static org.apache.hadoop.fs.permission.AclEntryType.GROUP;
+import static org.apache.hadoop.fs.permission.AclEntryType.MASK;
+import static org.apache.hadoop.fs.permission.AclEntryType.OTHER;
+import static org.apache.hadoop.fs.permission.AclEntryType.USER;
+import static org.apache.hadoop.fs.permission.FsAction.ALL;
+import static org.apache.hadoop.fs.permission.FsAction.NONE;
+import static org.apache.hadoop.fs.permission.FsAction.READ;
+import static org.apache.hadoop.fs.permission.FsAction.READ_EXECUTE;
+import static org.apache.hadoop.fs.permission.FsAction.READ_WRITE;
+import static org.apache.hadoop.hdfs.server.namenode.AclTestHelpers.aclEntry;
+import static org.junit.Assert.assertArrayEquals;
+
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.security.PrivilegedExceptionAction;
+import java.util.List;
 import java.util.Random;
 
 import org.apache.commons.logging.Log;
@@ -31,18 +47,27 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.TestDFSClientRetries;
+import org.apache.hadoop.hdfs.protocol.AclException;
 import org.apache.hadoop.hdfs.server.namenode.web.resources.NamenodeWebHdfsMethods;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.log4j.Level;
+import org.junit.AfterClass;
 import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
+import com.google.common.collect.Lists;
+
 /** Test WebHDFS */
 public class TestWebHDFS {
   static final Log LOG = LogFactory.getLog(TestWebHDFS.class);
@@ -51,6 +76,35 @@ public class TestWebHDFS {
   
   static final long systemStartTime = System.nanoTime();
 
+  private static MiniDFSCluster testCluster;
+  private static Configuration testConf;
+  private static FileSystem testFs;
+  private static int pathCount = 0;
+  private static Path tmpPath;
+
+  @BeforeClass
+  public static void init() throws Exception {
+    testConf = WebHdfsTestUtil.createConf();
+
+    testCluster = new MiniDFSCluster.Builder(testConf).numDataNodes(1).build();
+    testCluster.waitActive();
+    testFs = WebHdfsTestUtil.getWebHdfsFileSystem(testConf, WebHdfsFileSystem.SCHEME);
+  }
+
+  @AfterClass
+  public static void shutdown() throws Exception {
+    IOUtils.cleanup(null, testFs);
+    if (testCluster != null) {
+      testCluster.shutdown();
+    }
+  }
+
+  @Before
+  public void setUp() {
+    pathCount += 1;
+    tmpPath = new Path("/p" + pathCount);
+  }
+
   /** A timer for measuring performance. */
   static class Ticker {
     final String name;
@@ -300,4 +354,648 @@ public class TestWebHDFS {
     Assert.assertTrue(conf.getBoolean(DFSConfigKeys.DFS_WEBHDFS_ENABLED_KEY,
         false));
   }
+  
+  @Test
+  public void testModifyAclEntries() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE),
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+    
+    
+  }
+
+  @Test
+  public void testModifyAclEntriesOnlyAccess() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
+  }
+
+  @Test
+  public void testModifyAclEntriesOnlyDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testModifyAclEntriesMinimal() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo", READ_WRITE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ_WRITE),
+      aclEntry(ACCESS, GROUP, READ) }, returned);
+  }
+
+  @Test
+  public void testModifyAclEntriesMinimalDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testModifyAclEntriesCustomMask() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, MASK, NONE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ) }, returned);
+  }
+
+  @Test
+  public void testModifyAclEntriesStickyBit() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)01750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE),
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ_EXECUTE),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", READ_EXECUTE),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test(expected=FileNotFoundException.class)
+  public void testModifyAclEntriesPathNotFound() throws IOException {
+    // Path has not been created.
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+  }
+
+  @Test(expected=AclException.class)
+  public void testModifyAclEntriesDefaultOnFile() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.modifyAclEntries(tmpPath, aclSpec);
+  }
+
+  @Test
+  public void testRemoveAclEntries() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"),
+      aclEntry(DEFAULT, USER, "foo"));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testRemoveAclEntriesOnlyAccess() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, USER, "bar", READ_WRITE),
+      aclEntry(ACCESS, GROUP, READ_WRITE),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "bar", READ_WRITE),
+      aclEntry(ACCESS, GROUP, READ_WRITE) }, returned);
+  }
+
+  @Test
+  public void testRemoveAclEntriesOnlyDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, USER, "bar", READ_EXECUTE));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo"));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "bar", READ_EXECUTE),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testRemoveAclEntriesMinimal() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_WRITE),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"),
+      aclEntry(ACCESS, MASK));
+    System.out.println("AclSpec :"+aclSpec);
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+
+  @Test
+  public void testRemoveAclEntriesMinimalDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"),
+      aclEntry(ACCESS, MASK),
+      aclEntry(DEFAULT, USER, "foo"),
+      aclEntry(DEFAULT, MASK));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testRemoveAclEntriesStickyBit() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)01750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"),
+      aclEntry(DEFAULT, USER, "foo"));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test(expected=FileNotFoundException.class)
+  public void testRemoveAclEntriesPathNotFound() throws IOException {
+    // Path has not been created.
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, "foo"));
+    testFs.removeAclEntries(tmpPath, aclSpec);
+  }
+
+  @Test
+  public void testRemoveDefaultAcl() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeDefaultAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
+  }
+
+  @Test
+  public void testRemoveDefaultAclOnlyAccess() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeDefaultAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
+  }
+
+  @Test
+  public void testRemoveDefaultAclOnlyDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeDefaultAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test
+  public void testRemoveDefaultAclMinimal() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    testFs.removeDefaultAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test
+  public void testRemoveDefaultAclStickyBit() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)01750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeDefaultAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE) }, returned);
+  }
+
+  @Test(expected=FileNotFoundException.class)
+  public void testRemoveDefaultAclPathNotFound() throws IOException {
+    // Path has not been created.
+    testFs.removeDefaultAcl(tmpPath);
+  }
+
+  @Test
+  public void testRemoveAcl() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test
+  public void testRemoveAclMinimalAcl() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    testFs.removeAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test
+  public void testRemoveAclStickyBit() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)01750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.removeAcl(tmpPath);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test(expected=FileNotFoundException.class)
+  public void testRemoveAclPathNotFound() throws IOException {
+    // Path has not been created.
+    testFs.removeAcl(tmpPath);
+  }
+
+  @Test
+  public void testSetAcl() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, ALL),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testSetAclOnlyAccess() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ) }, returned);
+  }
+
+  @Test
+  public void testSetAclOnlyDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, ALL),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testSetAclMinimal() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] { }, returned);
+  }
+
+  @Test
+  public void testSetAclMinimalDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testSetAclCustomMask() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, MASK, ALL),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ) }, returned);
+  }
+
+  @Test
+  public void testSetAclStickyBit() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)01750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, ALL),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test(expected=FileNotFoundException.class)
+  public void testSetAclPathNotFound() throws IOException {
+    // Path has not been created.
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+  }
+
+  @Test(expected=AclException.class)
+  public void testSetAclDefaultOnFile() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+  }
+
+  @Test
+  public void testSetPermission() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short)0700));
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, ALL),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
+
+  @Test
+  public void testSetPermissionOnlyAccess() throws IOException {
+    testFs.create(tmpPath).close();
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short) 0640));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, READ_WRITE),
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ),
+      aclEntry(ACCESS, OTHER, NONE));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short)0600));
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(ACCESS, USER, "foo", READ),
+      aclEntry(ACCESS, GROUP, READ) }, returned);
+  }
+
+  @Test
+  public void testSetPermissionOnlyDefault() throws IOException {
+    FileSystem.mkdirs(testFs, tmpPath, FsPermission.createImmutable((short)0750));
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(ACCESS, USER, ALL),
+      aclEntry(ACCESS, GROUP, READ_EXECUTE),
+      aclEntry(ACCESS, OTHER, NONE),
+      aclEntry(DEFAULT, USER, "foo", ALL));
+    testFs.setAcl(tmpPath, aclSpec);
+    testFs.setPermission(tmpPath, FsPermission.createImmutable((short)0700));
+    AclStatus s = testFs.getAclStatus(tmpPath);
+    AclEntry[] returned = s.getEntries().toArray(new AclEntry[0]);
+    assertArrayEquals(new AclEntry[] {
+      aclEntry(DEFAULT, USER, ALL),
+      aclEntry(DEFAULT, USER, "foo", ALL),
+      aclEntry(DEFAULT, GROUP, READ_EXECUTE),
+      aclEntry(DEFAULT, MASK, ALL),
+      aclEntry(DEFAULT, OTHER, NONE) }, returned);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/resources/TestParam.java

@@ -21,12 +21,14 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
 import java.util.Arrays;
+import java.util.List;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.util.StringUtils;
@@ -300,4 +302,48 @@ public class TestParam {
 
     UserParam.setUserPatternDomain(oldDomain);
   }
+
+  @Test
+  public void testAclPermissionParam() {
+    final AclPermissionParam p =
+        new AclPermissionParam("user::rwx,group::r--,other::rwx,user:user1:rwx");
+    List<AclEntry> setAclList =
+        AclEntry.parseAclSpec("user::rwx,group::r--,other::rwx,user:user1:rwx",
+            true);
+    Assert.assertEquals(setAclList.toString(), p.getAclPermission(true)
+        .toString());
+
+    new AclPermissionParam("user::rw-,group::rwx,other::rw-,user:user1:rwx");
+    try {
+      new AclPermissionParam("user::rw--,group::rwx-,other::rw-");
+      Assert.fail();
+    } catch (IllegalArgumentException e) {
+      LOG.info("EXPECTED: " + e);
+    }
+
+    new AclPermissionParam(
+        "user::rw-,group::rwx,other::rw-,user:user1:rwx,group:group1:rwx,other::rwx,mask::rwx,default:user:user1:rwx");
+
+    try {
+      new AclPermissionParam("user:r-,group:rwx,other:rw-");
+      Assert.fail();
+    } catch (IllegalArgumentException e) {
+      LOG.info("EXPECTED: " + e);
+    }
+
+    try {
+      new AclPermissionParam("default:::r-,default:group::rwx,other::rw-");
+      Assert.fail();
+    } catch (IllegalArgumentException e) {
+      LOG.info("EXPECTED: " + e);
+    }
+
+    try {
+      new AclPermissionParam("user:r-,group::rwx,other:rw-,mask:rw-,temp::rwx");
+      Assert.fail();
+    } catch (IllegalArgumentException e) {
+      LOG.info("EXPECTED: " + e);
+    }
+  }
+ 
 }