HDFS-618. Support non-recursive mkdir(). Contributed by Kan Zhang

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/trunk@815918 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -157,6 +157,8 @@ Trunk (unreleased changes)
 
     HDFS-617. Support non-recursive create().  (Kan Zhang via szetszwo)
 
+    HDFS-618. Support non-recursive mkdir().  (Kan Zhang via szetszwo)
+
   BUG FIXES
 
     HDFS-76. Better error message to users when commands fail because of 

src/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -969,8 +969,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
 
   /**
    */
+  @Deprecated
   public boolean mkdirs(String src) throws IOException {
-    return mkdirs(src, null);
+    return mkdirs(src, null, true);
   }
 
   /**
@@ -980,10 +981,11 @@ public class DFSClient implements FSConstants, java.io.Closeable {
    * @param src The path of the directory being created
    * @param permission The permission of the directory being created.
    * If permission == null, use {@link FsPermission#getDefault()}.
+   * @param createParent create missing parent directory if true
    * @return True if the operation success.
-   * @see ClientProtocol#mkdirs(String, FsPermission)
+   * @see ClientProtocol#mkdirs(String, FsPermission, boolean)
    */
-  public boolean mkdirs(String src, FsPermission permission)throws IOException{
+  public boolean mkdirs(String src, FsPermission permission, boolean createParent)throws IOException{
     checkOpen();
     if (permission == null) {
       permission = FsPermission.getDefault();
@@ -991,11 +993,12 @@ public class DFSClient implements FSConstants, java.io.Closeable {
     FsPermission masked = permission.applyUMask(FsPermission.getUMask(conf));
     LOG.debug(src + ": masked=" + masked);
     try {
-      return namenode.mkdirs(src, masked);
+      return namenode.mkdirs(src, masked, createParent);
     } catch(RemoteException re) {
       throw re.unwrapRemoteException(AccessControlException.class,
                                      NSQuotaExceededException.class,
                                      DSQuotaExceededException.class,
+                                     FileNotFoundException.class,
                                      FileAlreadyExistsException.class);
     }
   }

src/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -280,9 +280,17 @@ public class DistributedFileSystem extends FileSystem {
     return stats;
   }
 
+  /**
+   * Create a directory with given name and permission, only when
+   * parent directory exists.
+   */
+  public boolean mkdir(Path f, FsPermission permission) throws IOException {
+    return dfs.mkdirs(getPathName(f), permission, false);
+  }
+
   @Override
   public boolean mkdirs(Path f, FsPermission permission) throws IOException {
-    return dfs.mkdirs(getPathName(f), permission);
+    return dfs.mkdirs(getPathName(f), permission, true);
   }
 
   /** {@inheritDoc} */

src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -44,9 +44,9 @@ public interface ClientProtocol extends VersionedProtocol {
    * Compared to the previous version the following changes have been introduced:
    * (Only the latest change is reflected.
    * The log of historical changes can be retrieved from the svn).
-   * 47: create() takes an additional boolean param createParent.
+   * 48: modified mkdirs() to take an additional boolean parameter
    */
-  public static final long versionID = 47L;
+  public static final long versionID = 48L;
   
   ///////////////////////////////////////
   // File contents
@@ -260,6 +260,7 @@ public interface ClientProtocol extends VersionedProtocol {
    *
    * @param src The path of the directory being created
    * @param masked The masked permission of the directory being created
+   * @param createParent create missing parent directory if true
    * @return True if the operation success.
    * @throws {@link AccessControlException} if permission to create file is 
    * denied by the system. As usually on the client side the exception will 
@@ -267,7 +268,8 @@ public interface ClientProtocol extends VersionedProtocol {
    * @throws QuotaExceededException if the operation would violate 
    *                                any quota restriction.
    */
-  public boolean mkdirs(String src, FsPermission masked) throws IOException;
+  public boolean mkdirs(String src, FsPermission masked, boolean createParent)
+      throws IOException;
 
   /**
    * Get a listing of the indicated directory

src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1520,9 +1520,9 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
   /**
    * Create all the necessary directories
    */
-  public boolean mkdirs(String src, PermissionStatus permissions
-      ) throws IOException {
-    boolean status = mkdirsInternal(src, permissions);
+  public boolean mkdirs(String src, PermissionStatus permissions,
+      boolean createParent) throws IOException {
+    boolean status = mkdirsInternal(src, permissions, createParent);
     getEditLog().logSync();
     if (status && auditLog.isInfoEnabled()) {
       final FileStatus stat = dir.getFileInfo(src);
@@ -1537,7 +1537,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
    * Create all the necessary directories
    */
   private synchronized boolean mkdirsInternal(String src,
-      PermissionStatus permissions) throws IOException {
+      PermissionStatus permissions, boolean createParent) throws IOException {
     NameNode.stateChangeLog.debug("DIR* NameSystem.mkdirs: " + src);
     if (isPermissionEnabled) {
       checkTraverse(src);
@@ -1556,6 +1556,10 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
       checkAncestorAccess(src, FsAction.WRITE);
     }
 
+    if (!createParent) {
+      verifyParentDir(src);
+    }
+
     // validate that we have enough inodes. This is, at best, a 
     // heuristic because the mkdirs() operation migth need to 
     // create multiple inodes.

src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -728,7 +728,7 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   }
     
   /** {@inheritDoc} */
-  public boolean mkdirs(String src, FsPermission masked) throws IOException {
+  public boolean mkdirs(String src, FsPermission masked, boolean createParent) throws IOException {
     stateChangeLog.debug("*DIR* NameNode.mkdirs: " + src);
     if (!checkPathLength(src)) {
       throw new IOException("mkdirs: Pathname too long.  Limit " 
@@ -736,7 +736,7 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     }
     return namesystem.mkdirs(src,
         new PermissionStatus(UserGroupInformation.getCurrentUGI().getUserName(),
-            null, masked));
+            null, masked), createParent);
   }
 
   /**

src/java/org/apache/hadoop/hdfs/server/namenode/NamenodeFsck.java

@@ -335,7 +335,7 @@ public class NamenodeFsck {
     String target = lostFound + file.getPath();
     String errmsg = "Failed to move " + file.getPath() + " to /lost+found";
     try {
-      if (!namenode.mkdirs(target, file.getPermission())) {
+      if (!namenode.mkdirs(target, file.getPermission(), true)) {
         LOG.warn(errmsg);
         return;
       }
@@ -501,7 +501,7 @@ public class NamenodeFsck {
       
       final FileStatus lfStatus = dfs.getFileInfo(lfName);
       if (lfStatus == null) { // not exists
-        lfInitedOk = dfs.mkdirs(lfName);
+        lfInitedOk = dfs.mkdirs(lfName, null, true);
         lostFound = lfName;
       } else if (!lfStatus.isDir()) { // exists but not a directory
         LOG.warn("Cannot use /lost+found : a regular file with this name exists.");

src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java

@@ -179,7 +179,7 @@ public class TestDFSClientRetries extends TestCase {
 
     public boolean delete(String src, boolean recursive) throws IOException { return false; }
 
-    public boolean mkdirs(String src, FsPermission masked) throws IOException { return false; }
+    public boolean mkdirs(String src, FsPermission masked, boolean createParent) throws IOException { return false; }
 
     public FileStatus[] getListing(String src) throws IOException { return null; }
 

src/test/hdfs/org/apache/hadoop/hdfs/TestDFSMkdirs.java

@@ -20,8 +20,10 @@ package org.apache.hadoop.hdfs;
 import junit.framework.TestCase;
 import java.io.*;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 
 
 /**
@@ -73,4 +75,46 @@ public class TestDFSMkdirs extends TestCase {
       cluster.shutdown();
     }
   }
+  
+  /**
+   * Tests mkdir will not create directory when parent is missing.
+   */
+  public void testMkdir() throws IOException {
+    Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
+    DistributedFileSystem dfs = (DistributedFileSystem) cluster.getFileSystem();
+    try {
+      // Create a dir in root dir, should succeed
+      assertTrue(dfs.mkdir(new Path("/mkdir-" + System.currentTimeMillis()),
+          FsPermission.getDefault()));
+      // Create a dir when parent dir exists as a file, should fail
+      IOException expectedException = null;
+      String filePath = "/mkdir-file-" + System.currentTimeMillis();
+      writeFile(dfs, new Path(filePath));
+      try {
+        dfs.mkdir(new Path(filePath + "/mkdir"), FsPermission.getDefault());
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Create a directory when parent dir exists as file using"
+          + " mkdir() should throw FileAlreadyExistsException ",
+          expectedException != null
+              && expectedException instanceof FileAlreadyExistsException);
+      // Create a dir in a non-exist directory, should fail
+      expectedException = null;
+      try {
+        dfs.mkdir(new Path("/non-exist/mkdir-" + System.currentTimeMillis()),
+            FsPermission.getDefault());
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Create a directory in a non-exist parent dir using"
+          + " mkdir() should throw FileNotFoundException ",
+          expectedException != null
+              && expectedException instanceof FileNotFoundException);
+    } finally {
+      dfs.close();
+      cluster.shutdown();
+    }
+  }
 }