[partial-ns] Implement SetPermission().

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

@@ -17,7 +17,9 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
+import com.google.protobuf.ByteString;
 import org.apache.hadoop.HadoopIllegalArgumentException;
+import org.apache.hadoop.fs.InvalidPathException;
 import org.apache.hadoop.fs.PathIsNotDirectoryException;
 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.fs.UnresolvedLinkException;
@@ -49,23 +51,23 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_ENABLED_KE
 
 public class FSDirAttrOp {
   static HdfsFileStatus setPermission(
-      FSDirectory fsd, final String srcArg, FsPermission permission)
+      FSDirectory fsd, final String src, FsPermission permission)
       throws IOException {
-    String src = srcArg;
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src);
-    INodesInPath iip;
-    fsd.writeLock();
-    try {
-      src = fsd.resolvePath(pc, src, pathComponents);
-      iip = fsd.getINodesInPath4Write(src);
+    try (RWTransaction tx = fsd.newRWTransaction().begin()) {
+      Resolver.Result paths = Resolver.resolve(tx, src);
+      if (paths.invalidPath()) {
+        throw new InvalidPathException(src);
+      } else if (paths.notFound()) {
+        throw new FileNotFoundException(src);
+      }
+      FlatINodesInPath iip = paths.inodesInPath();
       fsd.checkOwner(pc, iip);
-      unprotectedSetPermission(fsd, src, permission);
-    } finally {
-      fsd.writeUnlock();
+      unprotectedSetPermission(tx, src, permission);
+      tx.logSetPermissions(src, permission);
+      tx.commit();
+      return fsd.getAuditFileInfo(iip);
     }
-    fsd.getEditLog().logSetPermissions(src, permission);
-    return fsd.getAuditFileInfo(iip);
   }
 
   static HdfsFileStatus setOwner(
@@ -246,18 +248,18 @@ public class FSDirAttrOp {
     }
   }
 
-  static void unprotectedSetPermission(
-      FSDirectory fsd, String src, FsPermission permissions)
-      throws FileNotFoundException, UnresolvedLinkException,
-             QuotaExceededException, SnapshotAccessControlException {
-    assert fsd.hasWriteLock();
-    final INodesInPath inodesInPath = fsd.getINodesInPath4Write(src, true);
-    final INode inode = inodesInPath.getLastINode();
-    if (inode == null) {
-      throw new FileNotFoundException("File does not exist: " + src);
+  static void unprotectedSetPermission(RWTransaction tx, String src,
+      FsPermission permissions) throws IOException {
+    Resolver.Result paths = Resolver.resolve(tx, src);
+    if (paths.invalidPath()) {
+      throw new InvalidPathException(src);
+    } else if (paths.notFound()) {
+      throw new FileNotFoundException(src);
     }
-    int snapshotId = inodesInPath.getLatestSnapshotId();
-    inode.setPermission(permissions, snapshotId);
+    FlatINode inode = paths.inodesInPath().getLastINode();
+    ByteString b = new FlatINode.Builder().mergeFrom(inode).permission
+        (permissions.toShort()).build();
+    tx.putINode(inode.id(), b);
   }
 
   static void unprotectedSetOwner(

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

@@ -1563,6 +1563,11 @@ public class FSDirectory implements Closeable {
     checkPermission(pc, iip, true, null, null, null, null);
   }
 
+  void checkOwner(FSPermissionChecker pc, FlatINodesInPath iip)
+      throws AccessControlException {
+    // TODO
+  }
+
   void checkPathAccess(FSPermissionChecker pc, INodesInPath iip,
       FsAction access) throws AccessControlException {
     checkPermission(pc, iip, false, null, null, access, null);

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

@@ -581,8 +581,13 @@ public class FSEditLogLoader {
     }
     case OP_SET_PERMISSIONS: {
       SetPermissionsOp setPermissionsOp = (SetPermissionsOp)op;
-      FSDirAttrOp.unprotectedSetPermission(fsDir, renameReservedPathsOnUpgrade(
-          setPermissionsOp.src, logVersion), setPermissionsOp.permissions);
+      try (ReplayTransaction tx = fsDir.newReplayTransaction().begin()) {
+        String src = renameReservedPathsOnUpgrade(
+            setPermissionsOp.src, logVersion);
+        FSDirAttrOp.unprotectedSetPermission(tx, src, setPermissionsOp
+            .permissions);
+        tx.commit();
+      }
       break;
     }
     case OP_SET_OWNER: {

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

@@ -19,6 +19,7 @@ package org.apache.hadoop.hdfs.server.namenode;
 
 import com.google.protobuf.ByteString;
 import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.permission.FsPermission;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -167,4 +168,8 @@ class RWTransaction extends Transaction {
   public void logCloseFile(String path, FlatINode inode) {
     fsd.getEditLog().logCloseFile(fsd.ugid(), path, inode);
   }
+
+  public void logSetPermissions(String src, FsPermission permission) {
+    fsd.getEditLog().logSetPermissions(src, permission);
+  }
 }