From b053fdc547ddbb6322674142a14010683006d123 Mon Sep 17 00:00:00 2001
From: Mingliang Liu <liuml07@apache.org>
Date: Thu, 30 Mar 2017 13:03:34 -0700
Subject: [PATCH] HADOOP-14255. S3A to delete unnecessary fake directory
 objects in mkdirs(). Contributed by Mingliang Liu

---
 .../contract/AbstractContractMkdirTest.java   | 42 +++++++++++++++++++
 .../apache/hadoop/fs/s3a/S3AFileSystem.java   |  3 +-
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMkdirTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMkdirTest.java
index 427b0e972d..71d27067bd 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMkdirTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMkdirTest.java
@@ -126,4 +126,46 @@ public void testMkdirSlashHandling() throws Throwable {
     assertPathExists("check path existence without trailing slash failed",
         path("testmkdir/b"));
   }
+
+  @Test
+  public void testMkdirsPopulatingAllNonexistentAncestors() throws IOException {
+    describe("Verify mkdir will populate all its non-existent ancestors");
+    final FileSystem fs = getFileSystem();
+
+    final Path parent = path("testMkdirsPopulatingAllNonexistentAncestors");
+    assertTrue(fs.mkdirs(parent));
+    assertPathExists(parent + " should exist before making nested dir", parent);
+
+    Path nested = path(parent + "/a/b/c/d/e/f/g/h/i/j/k/L");
+    assertTrue(fs.mkdirs(nested));
+    while (nested != null && !nested.equals(parent) && !nested.isRoot()) {
+      assertPathExists(nested + " nested dir should exist", nested);
+      nested = nested.getParent();
+    }
+  }
+
+  @Test
+  public void testMkdirsDoesNotRemoveParentDirectories() throws IOException {
+    describe("Verify mkdir will make its parent existent");
+    final FileSystem fs = getFileSystem();
+
+    final Path parent = path("testMkdirsDoesNotRemoveParentDirectories");
+    assertTrue(fs.mkdirs(parent));
+
+    Path p = parent;
+    for (int i = 0; i < 10; i++) {
+      assertTrue(fs.mkdirs(p));
+      assertPathExists(p + " should exist after mkdir(" + p + ")", p);
+      p = path(p + "/dir-" + i);
+    }
+
+    // After mkdirs(sub-directory), its parent directory still exists
+    p = p.getParent();
+    while (p != null && !p.equals(parent) && !p.isRoot()) {
+      assertPathExists("Path " + p + " should exist", p);
+      assertIsDirectory(p);
+      p = p.getParent();
+    }
+  }
+
 }
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index b17281be8f..9eb55756cd 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -1525,8 +1525,6 @@ public boolean mkdirs(Path path, FsPermission permission) throws IOException,
    * @throws IOException other IO problems
    * @throws AmazonClientException on failures inside the AWS SDK
    */
-  // TODO: If we have created an empty file at /foo/bar and we then call
-  // mkdirs for /foo/bar/baz/roo what happens to the empty file /foo/bar/?
   private boolean innerMkdirs(Path f, FsPermission permission)
       throws IOException, FileAlreadyExistsException, AmazonClientException {
     LOG.debug("Making directory: {}", f);
@@ -1561,6 +1559,7 @@ private boolean innerMkdirs(Path f, FsPermission permission)
 
       String key = pathToKey(f);
       createFakeDirectory(key);
+      deleteUnnecessaryFakeDirectories(f.getParent());
       return true;
     }
   }