Merged r1197856 from branch-0.20-security for HDFS-617.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-205@1197860 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -15,12 +15,15 @@ Release 0.20.205.1 - unreleased
     HDFS-2450. Filesystem supports path with both short names and FQDN.
     (Daryn Sharp via suresh)
 
+    HDFS-617. Support for non-recursive create() in HDFS.
+    (Kan Zhang via jitendra)
 
   BUG FIXES
 
     HADOOP-7784. Fixed jsvc packaging. (Eric Yang)
 
-    HADOOP-7740. Fixed security audit logger configuration. (Arpit Gupta via Eric Yang)
+    HADOOP-7740. Fixed security audit logger configuration. 
+    (Arpit Gupta via Eric Yang)
 
     HADOOP-7765. Clean packaging working directory for Debian packaging.
     (Eric Yang)

src/core/org/apache/hadoop/fs/FileAlreadyExistsException.java

@@ -0,0 +1,39 @@
+/**
+ * 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.fs;
+
+import java.io.IOException;
+
+/**
+ * Used when target file already exists for any operation and 
+ * is not configured to be overwritten.  
+ */
+public class FileAlreadyExistsException
+    extends IOException {
+
+  private static final long serialVersionUID = 1L;
+
+  public FileAlreadyExistsException() {
+    super();
+  }
+
+  public FileAlreadyExistsException(String msg) {
+    super(msg);
+  }
+}

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

@@ -24,6 +24,7 @@ import org.apache.hadoop.io.retry.RetryProxy;
 import org.apache.hadoop.fs.*;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.ipc.*;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.*;
@@ -563,6 +564,23 @@ public class DFSClient implements FSConstants, java.io.Closeable {
     return create(src, FsPermission.getDefault(),
         overwrite, replication, blockSize, progress, buffersize);
   }
+  /**
+   * Call
+   * {@link #create(String,FsPermission,boolean,boolean,short,long,Progressable,int)}
+   * with createParent set to true.
+   */
+  public OutputStream create(String src, 
+      FsPermission permission,
+      boolean overwrite,
+      short replication,
+      long blockSize,
+      Progressable progress,
+      int buffersize
+      ) throws IOException {
+    return create(src, permission, overwrite, true,
+        replication, blockSize, progress, buffersize);
+  }
+
   /**
    * Create a new dfs file with the specified block replication 
    * with write-progress reporting and return an output stream for writing
@@ -572,6 +590,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
    * @param permission The permission of the directory being created.
    * If permission == null, use {@link FsPermission#getDefault()}.
    * @param overwrite do not check for file existence if true
+   * @param createParent create missing parent directory if true
    * @param replication block replication
    * @return output stream
    * @throws IOException
@@ -580,6 +599,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
   public OutputStream create(String src, 
                              FsPermission permission,
                              boolean overwrite, 
+                             boolean createParent,
                              short replication,
                              long blockSize,
                              Progressable progress,
@@ -592,7 +612,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
     FsPermission masked = permission.applyUMask(FsPermission.getUMask(conf));
     LOG.debug(src + ": masked=" + masked);
     OutputStream result = new DFSOutputStream(src, masked,
-        overwrite, replication, blockSize, progress, buffersize,
+        overwrite, createParent, replication, blockSize, progress, buffersize,
         conf.getInt("io.bytes.per.checksum", 512));
     leasechecker.put(src, result);
     return result;
@@ -3060,7 +3080,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
      * @see ClientProtocol#create(String, FsPermission, String, boolean, short, long)
      */
     DFSOutputStream(String src, FsPermission masked, boolean overwrite,
-        short replication, long blockSize, Progressable progress,
+        boolean createParent, short replication, long blockSize, Progressable progress,
         int buffersize, int bytesPerChecksum) throws IOException {
       this(src, blockSize, progress, bytesPerChecksum, replication);
 
@@ -3068,9 +3088,11 @@ public class DFSClient implements FSConstants, java.io.Closeable {
 
       try {
         namenode.create(
-            src, masked, clientName, overwrite, replication, blockSize);
+            src, masked, clientName, overwrite, createParent, replication, blockSize);
       } catch(RemoteException re) {
         throw re.unwrapRemoteException(AccessControlException.class,
+                                       FileAlreadyExistsException.class,
+                                       FileNotFoundException.class,
                                        NSQuotaExceededException.class,
                                        DSQuotaExceededException.class);
       }

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

@@ -181,10 +181,25 @@ public class DistributedFileSystem extends FileSystem {
     statistics.incrementWriteOps(1);
     return new FSDataOutputStream
        (dfs.create(getPathName(f), permission,
-                   overwrite, replication, blockSize, progress, bufferSize),
+                   overwrite, true, replication, blockSize, progress, bufferSize),
         statistics);
   }
 
+  /**
+   * Same as create(), except fails if parent directory doesn't already exist.
+   * @see #create(Path, FsPermission, boolean, int, short, long, Progressable)
+   */
+  public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
+      boolean overwrite,
+      int bufferSize, short replication, long blockSize, 
+      Progressable progress) throws IOException {
+
+    return new FSDataOutputStream
+        (dfs.create(getPathName(f), permission, 
+                    overwrite, false, replication, blockSize, progress, bufferSize), 
+         statistics);
+  }
+
   public boolean setReplication(Path src, 
                                 short replication
                                ) throws IOException {

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

@@ -103,6 +103,7 @@ public interface ClientProtocol extends VersionedProtocol {
    * @param clientName name of the current client.
    * @param overwrite indicates whether the file should be 
    * overwritten if it already exists.
+   * @param createParent create missing parent directory if true
    * @param replication block replication factor.
    * @param blockSize maximum block size.
    * 
@@ -117,10 +118,22 @@ public interface ClientProtocol extends VersionedProtocol {
                      FsPermission masked,
                              String clientName, 
                              boolean overwrite, 
+                             boolean createParent,
                              short replication,
                              long blockSize
                              ) throws IOException;
 
+  /**
+   * Create a new file entry in the namespace.
+   * 
+   */
+  public void create(String src, 
+                     FsPermission masked,
+                             String clientName, 
+                             boolean overwrite, 
+                             short replication,
+                             long blockSize
+                             ) throws IOException;
   /**
    * Append to the end of the file. 
    * @param src path of the file being created.

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

@@ -100,6 +100,7 @@ import org.apache.hadoop.hdfs.server.protocol.BalancerBandwidthCommand;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.metrics2.MetricsBuilder;
 import org.apache.hadoop.metrics2.MetricsSource;
 import org.apache.hadoop.metrics2.MetricsSystem;
@@ -1151,6 +1152,24 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
                             text + " is less than the required minimum " + minReplication);
   }
 
+  /*
+   * Verify that parent dir exists
+   */
+  private void verifyParentDir(String src) throws FileAlreadyExistsException,
+      FileNotFoundException {
+    Path parent = new Path(src).getParent();
+    if (parent != null) {
+      INode[] pathINodes = dir.getExistingPathINodes(parent.toString());
+      if (pathINodes[pathINodes.length - 1] == null) {
+        throw new FileNotFoundException("Parent directory doesn't exist: "
+            + parent.toString());
+      } else if (!pathINodes[pathINodes.length - 1].isDirectory()) {
+        throw new FileAlreadyExistsException("Parent path is not a directory: "
+            + parent.toString());
+      }
+    }
+  }
+
   /**
    * Create a new file entry in the namespace.
    * 
@@ -1161,10 +1180,10 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
    */
   void startFile(String src, PermissionStatus permissions,
                  String holder, String clientMachine,
-                 boolean overwrite, short replication, long blockSize
+                 boolean overwrite, boolean createParent, short replication, long blockSize
                 ) throws IOException {
     startFileInternal(src, permissions, holder, clientMachine, overwrite, false,
-                      replication, blockSize);
+                      createParent, replication, blockSize);
     getEditLog().logSync();
     if (auditLog.isInfoEnabled() && isExternalInvocation()) {
       final HdfsFileStatus stat = dir.getFileInfo(src);
@@ -1180,6 +1199,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
                                               String clientMachine, 
                                               boolean overwrite,
                                               boolean append,
+                                              boolean createParent,
                                               short replication,
                                               long blockSize
                                               ) throws IOException {
@@ -1187,6 +1207,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
       NameNode.stateChangeLog.debug("DIR* NameSystem.startFile: src=" + src
           + ", holder=" + holder
           + ", clientMachine=" + clientMachine
+          + ", createParent=" + createParent
           + ", replication=" + replication
           + ", overwrite=" + overwrite
           + ", append=" + append);
@@ -1213,6 +1234,10 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
       }
     }
 
+    if (!createParent) {
+      verifyParentDir(src);
+    }
+
     try {
       INode myFile = dir.getFileINode(src);
       recoverLeaseInternal(myFile, src, holder, clientMachine, false);
@@ -1396,7 +1421,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
                             " Please refer to dfs.support.append configuration parameter.");
     }
     startFileInternal(src, null, holder, clientMachine, false, true, 
-                      (short)maxReplication, (long)0);
+                      false, (short)maxReplication, (long)0);
     getEditLog().logSync();
 
     //

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

@@ -597,11 +597,23 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     return clientMachine;
   }
 
+  @Deprecated
+  public void create(String src, 
+                     FsPermission masked,
+                             String clientName, 
+                             boolean overwrite,
+                             short replication,
+                             long blockSize
+                             ) throws IOException {
+    create(src,masked,clientName,overwrite,true,replication,blockSize);
+  }
+
   /** {@inheritDoc} */
   public void create(String src, 
                      FsPermission masked,
                              String clientName, 
                              boolean overwrite,
+                             boolean createParent,
                              short replication,
                              long blockSize
                              ) throws IOException {
@@ -617,7 +629,7 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     namesystem.startFile(src,
         new PermissionStatus(UserGroupInformation.getCurrentUser().getShortUserName(),
             null, masked),
-        clientName, clientMachine, overwrite, replication, blockSize);
+        clientName, clientMachine, overwrite, createParent, replication, blockSize);
     myMetrics.incrNumFilesCreated();
     myMetrics.incrNumCreateFileOps();
   }

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

@@ -227,8 +227,11 @@ public class TestDFSClientRetries extends TestCase {
     // The following methods are stub methods that are not needed by this mock class
     public LocatedBlocks  getBlockLocations(String src, long offset, long length) throws IOException { return null; }
 
+    @Deprecated
     public void create(String src, FsPermission masked, String clientName, boolean overwrite, short replication, long blockSize) throws IOException {}
 
+    public void create(String src, FsPermission masked, String clientName, boolean overwrite, boolean createparent, short replication, long blockSize) throws IOException {}
+
     public LocatedBlock append(String src, String clientName) throws IOException { return null; }
 
     public boolean setReplication(String src, short replication) throws IOException { return false; }

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

@@ -19,6 +19,7 @@ package org.apache.hadoop.hdfs;
 
 import java.io.BufferedReader;
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.FileReader;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -28,9 +29,11 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileStatus;
 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.Block;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
@@ -309,6 +312,98 @@ public class TestFileCreation extends junit.framework.TestCase {
     }
   }
 
+  /**
+   * Test file creation using createNonRecursive().
+   */
+  public void testFileCreationNonRecursive() throws IOException {
+    Configuration conf = new Configuration();
+    if (simulatedStorage) {
+      conf.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
+    }
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 1, true, null);
+    FileSystem fs = cluster.getFileSystem();
+    final Path path = new Path("/" + System.currentTimeMillis()
+        + "-testFileCreationNonRecursive");
+    FSDataOutputStream out = null;
+
+    try {
+      IOException expectedException = null;
+      final String nonExistDir = "/non-exist-" + System.currentTimeMillis();
+
+      fs.delete(new Path(nonExistDir), true);
+      // Create a new file in root dir, should succeed
+      out = createNonRecursive(fs, path, 1, false);
+      out.close();
+      // Create a file when parent dir exists as file, should fail
+      expectedException = null;
+      try {
+        createNonRecursive(fs, new Path(path, "Create"), 1, false);
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Create a file when parent directory exists as a file"
+          + " should throw FileAlreadyExistsException ",
+          expectedException != null
+              && expectedException instanceof FileAlreadyExistsException);
+      fs.delete(path, true);
+      // Create a file in a non-exist directory, should fail
+      final Path path2 = new Path(nonExistDir + "/testCreateNonRecursive");
+      expectedException = null;
+      try {
+        createNonRecursive(fs, path2, 1, false);
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Create a file in a non-exist dir using"
+          + " createNonRecursive() should throw FileNotFoundException ",
+          expectedException != null
+              && expectedException instanceof FileNotFoundException);
+
+      // Overwrite a file in root dir, should succeed
+      out = createNonRecursive(fs, path, 1, true);
+      out.close();
+      // Overwrite a file when parent dir exists as file, should fail
+      expectedException = null;
+      try {
+        createNonRecursive(fs, new Path(path, "Overwrite"), 1, true);
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Overwrite a file when parent directory exists as a file"
+          + " should throw FileAlreadyExistsException ",
+          expectedException != null
+              && expectedException instanceof FileAlreadyExistsException);
+      fs.delete(path, true);
+      // Overwrite a file in a non-exist directory, should fail
+      final Path path3 = new Path(nonExistDir + "/testOverwriteNonRecursive");
+      expectedException = null;
+      try {
+        createNonRecursive(fs, path3, 1, true);
+      } catch (IOException e) {
+        expectedException = e;
+      }
+      assertTrue("Overwrite a file in a non-exist dir using"
+          + " createNonRecursive() should throw FileNotFoundException ",
+          expectedException != null
+              && expectedException instanceof FileNotFoundException);
+    } finally {
+      fs.close();
+      cluster.shutdown();
+    }
+  }
+
+  // creates a file using DistributedFileSystem.createNonRecursive()
+  static FSDataOutputStream createNonRecursive(FileSystem fs, Path name,
+      int repl, boolean overwrite) throws IOException {
+    System.out.println("createNonRecursive: Created " + name + " with " + repl
+        + " replica.");
+    FSDataOutputStream stm = ((DistributedFileSystem) fs).createNonRecursive(
+        name, FsPermission.getDefault(), overwrite, fs.getConf().getInt(
+            "io.file.buffer.size", 4096), (short) repl, (long) blockSize, null);
+
+    return stm;
+  }
+  
   /**
    * Test that file data does not become corrupted even in the face of errors.
    */

src/test/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java

@@ -546,7 +546,7 @@ public class NNThroughputBenchmark {
       long start = System.currentTimeMillis();
       // dummyActionNoSynch(fileIdx);
       nameNode.create(fileNames[daemonId][inputIdx], FsPermission.getDefault(),
-                      clientName, true, replication, BLOCK_SIZE);
+                      clientName, true, true, replication, BLOCK_SIZE);
       long end = System.currentTimeMillis();
       for(boolean written = !closeUponCreate; !written; 
         written = nameNode.complete(fileNames[daemonId][inputIdx], clientName));
@@ -917,7 +917,7 @@ public class NNThroughputBenchmark {
       for(int idx=0; idx < nrFiles; idx++) {
         String fileName = nameGenerator.getNextFileName("ThroughputBench");
         nameNode.create(fileName, FsPermission.getDefault(),
-                        clientName, true, replication, BLOCK_SIZE);
+                        clientName, true, true, replication, BLOCK_SIZE);
         addBlocks(fileName, clientName);
         nameNode.complete(fileName, clientName);
       }