From 587061103097160d8aceb60dbef6958cafdd30ae Mon Sep 17 00:00:00 2001
From: Steve Loughran <stevel@apache.org>
Date: Wed, 1 Jun 2016 14:17:18 +0100
Subject: [PATCH] HADOOP-13162. Consider reducing number of getFileStatus calls
 in S3AFileSystem.mkdirs. (Rajesh Balamohan via stevel)

---
 .../fs/FileContextCreateMkdirBaseTest.java    | 73 ++++++++++++++++++-
 .../apache/hadoop/fs/s3a/S3AFileSystem.java   |  3 +
 2 files changed, 73 insertions(+), 3 deletions(-)

diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextCreateMkdirBaseTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextCreateMkdirBaseTest.java
index d91091f07c..c1de27a347 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextCreateMkdirBaseTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextCreateMkdirBaseTest.java
@@ -20,14 +20,15 @@
 
 import java.io.IOException;
 
-import org.apache.hadoop.util.StringUtils;
 import org.apache.log4j.Level;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import static org.apache.hadoop.fs.FileContextTestHelper.*;
-import org.apache.commons.logging.impl.Log4JLogger;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.assertIsDirectory;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.assertIsFile;
+
 import org.apache.hadoop.test.GenericTestUtils;
 
 /**
@@ -116,7 +117,73 @@ public void testMkdirRecursiveWithNonExistingDir() throws IOException {
     fc.mkdir(f, FileContext.DEFAULT_PERM, true);
     Assert.assertTrue(isDir(fc, f));
   }
- 
+
+  @Test
+  public void testMkdirsRecursiveWithExistingDir() throws IOException {
+    Path f = getTestRootPath(fc, "aDir/bDir/cDir");
+    fc.mkdir(f, FileContext.DEFAULT_PERM, true);
+    assertIsDirectory(fc.getFileStatus(f));
+    assertIsDirectory(fc.getFileStatus(f.getParent()));
+    assertIsDirectory(fc.getFileStatus(f.getParent().getParent()));
+  }
+
+  @Test
+  public void testMkdirRecursiveWithExistingFile() throws IOException {
+    Path f = getTestRootPath(fc, "NonExistant3/aDir");
+    fc.mkdir(f, FileContext.DEFAULT_PERM, true);
+    assertIsDirectory(fc.getFileStatus(f));
+    assertIsDirectory(fc.getFileStatus(f.getParent()));
+
+    // create a sample file
+    Path filePath = new Path(f.getParent(), "test.txt");
+    createFile(fc, filePath);
+    assertIsFile(filePath, fc.getFileStatus(filePath));
+
+    // try creating another folder which conflicts with filePath
+    Path dirPath = new Path(filePath, "bDir/cDir");
+    try {
+      fc.mkdir(dirPath, FileContext.DEFAULT_PERM, true);
+      Assert.fail("Mkdir for " + dirPath
+          + " should have failed as a file was present");
+    } catch(IOException e) {
+      // failed as expected
+    }
+  }
+
+  @Test
+  public void testWithRename() throws IOException, InterruptedException {
+    Path root = getTestRootPath(fc);
+    Path f = new Path(root, "d1/d2/d3");
+    fc.mkdir(f, FileContext.DEFAULT_PERM, true);
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2/d3")));
+
+    // create a sample file f.txt
+    Path fPath = new Path(root, "d1/d2/f.txt");
+    createFile(fc, fPath);
+    assertIsFile(fPath, fc.getFileStatus(fPath));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2/d3")));
+
+    // create a sample file f2.txt
+    Path f2Path = new Path(getTestRootPath(fc), "d1/d2/d3/f2.txt");
+    createFile(fc, f2Path);
+    assertIsFile(fPath, fc.getFileStatus(f2Path));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d2/d3")));
+
+    //rename d1/d2/d3 d1/d4
+    fc.rename(new Path(root, "d1/d2/d3"), new Path(root, "d1/d4"));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1")));
+    assertIsDirectory(fc.getFileStatus(new Path(root, "d1/d4")));
+    Path f2NewPath = new Path(root, "d1/d4/f2.txt");
+    assertIsFile(f2NewPath, fc.getFileStatus(f2NewPath));
+  }
+
+
   ///////////////////////
   //      Test Create
   ////////////////////////
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 0eb720af3e..581518ac7f 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
@@ -1088,6 +1088,9 @@ private boolean innerMkdirs(Path f, FsPermission permission)
       do {
         try {
           FileStatus fileStatus = getFileStatus(fPart);
+          if (fileStatus.isDirectory()) {
+            break;
+          }
           if (fileStatus.isFile()) {
             throw new FileAlreadyExistsException(String.format(
                 "Can't make directory for path '%s' since it is a file.",