HADOOP-14428. s3a: mkdir appears to be broken. Contributed by Mingliang Liu

(cherry picked from commit 6aeda55bb8f741d9dafd41f6dfbf1a88acdd4003)

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMkdirTest.java

@@ -113,18 +113,25 @@ public abstract class AbstractContractMkdirTest extends AbstractFSContractTestBa
     describe("verify mkdir slash handling");
     FileSystem fs = getFileSystem();
 
-    // No trailing slash
-    assertTrue(fs.mkdirs(path("testmkdir/a")));
-    assertPathExists("mkdir without trailing slash failed",
-        path("testmkdir/a"));
-
-    // With trailing slash
-    assertTrue(fs.mkdirs(path("testmkdir/b/")));
-    assertPathExists("mkdir with trailing slash failed", path("testmkdir/b/"));
-
-    // Mismatched slashes
-    assertPathExists("check path existence without trailing slash failed",
-        path("testmkdir/b"));
+    final Path[] paths = new Path[] {
+        path("testMkdirSlashHandling/a"), // w/o trailing slash
+        path("testMkdirSlashHandling/b/"), // w/ trailing slash
+        // unqualified w/o trailing slash
+        new Path(getContract().getTestPath() + "/testMkdirSlashHandling/c"),
+        // unqualified w/ trailing slash
+        new Path(getContract().getTestPath() + "/testMkdirSlashHandling/d/"),
+        // unqualified w/ multiple trailing slashes
+        new Path(getContract().getTestPath() + "/testMkdirSlashHandling/e///")
+    };
+    for (Path path : paths) {
+      assertTrue(fs.mkdirs(path));
+      assertPathExists(path + " does not exist after mkdirs", path);
+      assertIsDirectory(path);
+      if (path.toString().endsWith("/")) {
+        String s = path.toString().substring(0, path.toString().length() - 1);
+        assertIsDirectory(new Path(s));
+      }
+    }
   }
 
   @Test

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -1586,7 +1586,9 @@ public class S3AFileSystem extends FileSystem {
 
       String key = pathToKey(f);
       createFakeDirectory(key);
-      deleteUnnecessaryFakeDirectories(f.getParent());
+      // this is complicated because getParent(a/b/c/) returns a/b/c, but
+      // we want a/b. See HADOOP-14428 for more details.
+      deleteUnnecessaryFakeDirectories(new Path(f.toString()).getParent());
       return true;
     }
   }
@@ -1955,6 +1957,7 @@ public class S3AFileSystem extends FileSystem {
     while (!path.isRoot()) {
       String key = pathToKey(path);
       key = (key.endsWith("/")) ? key : (key + "/");
+      LOG.trace("To delete unnecessary fake directory {} for {}", key, path);
       keysToRemove.add(new DeleteObjectsRequest.KeyVersion(key));
       path = path.getParent();
     }