HADOOP-240. Fix DFS mkdirs() to not warn when directories already exist. Contributed by Hairong.

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

CHANGES.txt

@@ -14,6 +14,9 @@ Trunk (unreleased changes)
     test to be run when "DistributedFSCheck" was specified.
    (Konstantin Shvachko via cutting)
 
+ 4. HADOOP-240.  DFS's mkdirs() implementation no longer logs a warning
+    when the directory already exists. (Hairong Kuang via cutting)
+
 
 Release 0.3.1 - 2006-06-05
 

src/java/org/apache/hadoop/dfs/FSDirectory.java

@@ -143,18 +143,22 @@ class FSDirectory implements FSConstants {
          * @param path file path
          * @param newNode INode to be added
          * @return null if the node already exists; inserted INode, otherwise
+         * @throws FileNotFoundException 
          * @author shv
          */
-        INode addNode(String path, INode newNode) {
+        INode addNode(String path, INode newNode) throws FileNotFoundException {
           File target = new File( path );
           // find parent
           Path parent = new Path(path).getParent();
-          if (parent == null)
-            return null;
+          if (parent == null) { // add root
+              return null;
+          }
           INode parentNode = getNode(parent.toString());
-          if (parentNode == null)
-            return null;
-          // check whether the parent already has a node with that name
+          if (parentNode == null) {
+              throw new FileNotFoundException(
+                      "Parent path does not exist: "+path);
+          }
+           // check whether the parent already has a node with that name
           String name = newNode.name = target.getName();
           if( parentNode.getChild( name ) != null )
             return null;
@@ -688,11 +692,19 @@ class FSDirectory implements FSConstants {
      */
     boolean unprotectedAddFile(UTF8 path, INode newNode) {
       synchronized (rootDir) {
-        int nrBlocks = (newNode.blocks == null) ? 0 : newNode.blocks.length;
-        // Add file->block mapping
-        for (int i = 0; i < nrBlocks; i++)
-            activeBlocks.put(newNode.blocks[i], newNode);
-        return (rootDir.addNode(path.toString(), newNode) != null);
+         try {
+            if( rootDir.addNode(path.toString(), newNode ) != null ) {
+                int nrBlocks = (newNode.blocks == null) ? 0 : newNode.blocks.length;
+                // Add file->block mapping
+                for (int i = 0; i < nrBlocks; i++)
+                    activeBlocks.put(newNode.blocks[i], newNode);
+                return true;
+            } else {
+                return false;
+            }
+        } catch (FileNotFoundException e ) {
+            return false;
+        }
       }
     }
 
@@ -720,23 +732,36 @@ class FSDirectory implements FSConstants {
             INode renamedNode = rootDir.getNode(srcStr);
             if (renamedNode == null) {
                 NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-                        +"failed to rename "+src+" to "+dst+ " because "+ src+" does not exist" );
+                        +"failed to rename "+src+" to "+dst+ " because source does not exist" );
                 return false;
             }
-            renamedNode.removeNode();
             if (isDir(dst)) {
               dstStr += "/" + new File(srcStr).getName();
             }
+            if( rootDir.getNode(dstStr.toString()) != null ) {
+                NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+                        +"failed to rename "+src+" to "+dstStr+ " because destination exists" );
+                return false;
+            }
+            renamedNode.removeNode();
+            
             // the renamed node can be reused now
-            if( rootDir.addNode(dstStr, renamedNode ) == null ) {
+            try {
+                if( rootDir.addNode(dstStr, renamedNode ) != null ) {
+                    NameNode.stateChangeLog.debug("DIR* FSDirectory.unprotectedRenameTo: "
+                        +src+" is renamed to "+dst );
+                    return true;
+                }
+            } catch (FileNotFoundException e ) {
                 NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
                         +"failed to rename "+src+" to "+dst );
-              rootDir.addNode(srcStr, renamedNode); // put it back
-              return false;
+                try {
+                    rootDir.addNode(srcStr, renamedNode); // put it back
+                }catch(FileNotFoundException e2) {                
+                }
             }
-            NameNode.stateChangeLog.debug("DIR* FSDirectory.unprotectedRenameTo: "
-                     +src+" is renamed to "+dst );
-            return true;
+
+            return false;
         }
     }
 
@@ -977,29 +1002,28 @@ class FSDirectory implements FSConstants {
 
         // Now go backwards through list of dirs, creating along
         // the way
-        boolean lastSuccess = false;
         int numElts = v.size();
         for (int i = numElts - 1; i >= 0; i--) {
             String cur = (String) v.elementAt(i);
-            INode inserted = unprotectedMkdir(cur);
-            if (inserted != null) {
-                NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
+            try {
+               INode inserted = unprotectedMkdir(cur);
+               if (inserted != null) {
+                   NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
                         +"created directory "+cur );
-                logEdit(OP_MKDIR, new UTF8(inserted.computeName()), null);
-                lastSuccess = true;
-            } else {
-                lastSuccess = false;
+                   logEdit(OP_MKDIR, new UTF8(inserted.computeName()), null);
+               } // otherwise cur exists, continue
+            } catch (FileNotFoundException e ) {
+                NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
+                        +"failed to create directory "+src);
+                return false;
             }
         }
-/*        if( !lastSuccess )
-            NameNode.stateChangeLog.warn("DIR* FSDirectory.mkdirs: "
-                    +"failed to create directory "+src );*/
-        return lastSuccess;
+        return true;
     }
 
     /**
      */
-    INode unprotectedMkdir(String src) {
+    INode unprotectedMkdir(String src) throws FileNotFoundException {
         synchronized (rootDir) {
             return rootDir.addNode(src, new INode(new File(src).getName()));
         }

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -1266,7 +1266,7 @@ class FSNamesystem implements FSConstants {
 
             if (! dir.isValidBlock(b) && ! pendingCreateBlocks.contains(b)) {
                 obsolete.add(b);
-                NameNode.stateChangeLog.info("BLOCK* NameSystem.processReport: "
+                NameNode.stateChangeLog.debug("BLOCK* NameSystem.processReport: "
                         +"ask "+nodeID.getName()+" to delete "+b.getBlockName() );
             }
         }
@@ -1509,7 +1509,7 @@ class FSNamesystem implements FSConstants {
                 blockList.append(' ');
                 blockList.append(((Block)invalidateSet.elementAt(i)).getBlockName());
             }
-            NameNode.stateChangeLog.info("BLOCK* NameSystem.blockToInvalidate: "
+            NameNode.stateChangeLog.debug("BLOCK* NameSystem.blockToInvalidate: "
                    +"ask "+nodeID.getName()+" to delete " + blockList );
         }
         return (Block[]) invalidateSet.toArray(new Block[invalidateSet.size()]);

src/test/org/apache/hadoop/dfs/ClusterTestDFSNamespaceLogging.java

@@ -73,7 +73,7 @@ public class ClusterTestDFSNamespaceLogging extends TestCase implements FSConsta
   private static final int BLOCK_LOG_HEADER_LEN = 32;
   /** DFS block size
    */
-  private static final int BLOCK_SIZE = 32*1000*1000;
+  private static final int BLOCK_SIZE = 32*1024*1024;
   
   /** Buffer size
    */
@@ -158,7 +158,7 @@ public class ClusterTestDFSNamespaceLogging extends TestCase implements FSConsta
     
       // create a file with 2 data blocks
       try {
-        createFile("/data/yy",BLOCK_SIZE+1);
+        createFile("/data/yy", BLOCK_SIZE+1);
         assertCreate( "/data/yy", BLOCK_SIZE+1, false );
       } catch( IOException ioe ) {
     	assertCreate( "/data/yy", BLOCK_SIZE+1, true );
@@ -326,9 +326,9 @@ public class ClusterTestDFSNamespaceLogging extends TestCase implements FSConsta
   //
   private void configureDFS() throws IOException {
 	// set given config param to override other config settings
-	conf.setInt("test.dfs.block_size", BLOCK_SIZE);
+	conf.setInt("dfs.block.size", BLOCK_SIZE);
 	// verify that config changed
-	assertTrue(BLOCK_SIZE == conf.getInt("test.dfs.block_size", 2)); // 2 is an intentional obviously-wrong block size
+	assertTrue(BLOCK_SIZE == conf.getInt("dfs.block.size", 2)); // 2 is an intentional obviously-wrong block size
 	// downsize for testing (just to save resources)
 	conf.setInt("dfs.namenode.handler.count", 3);
 	conf.setLong("dfs.blockreport.intervalMsec", 50*1000L);