HADOOP-18544: S3A: add option to disable probe for dir marker recreation on delete/rename.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/AbstractFileSystem.java

@@ -850,8 +850,23 @@ public void renameInternal(final Path src, final Path dst,
       FileAlreadyExistsException, FileNotFoundException,
       ParentNotDirectoryException, UnresolvedLinkException, IOException {
     // Default implementation deals with overwrite in a non-atomic way
-    final FileStatus srcStatus = getFileLinkStatus(src);
+    if (checkIfCanRename(src, dst, overwrite)) {
+      delete(dst, false);
+    }
+    renameInternal(src, dst);
+  }
 
+  /**
+   * Check if the given rename can proceed and return true if the destination directory or file needs
+   * to be deleted for the rename to proceed.
+   * @param src source of the file
+   * @param dst dst of the rename
+   * @param overwrite if overwrite is allowed
+   * @return a boolean reflecting if the dst needs to be deleted
+   * @throws IOException
+   */
+  public boolean checkIfCanRename(Path src, Path dst, boolean overwrite) throws IOException {
+    final FileStatus srcStatus = getFileLinkStatus(src);
     FileStatus dstStatus;
     try {
       dstStatus = getFileLinkStatus(dst);
@@ -884,16 +899,15 @@ public void renameInternal(final Path src, final Path dst,
               "Rename cannot overwrite non empty destination directory " + dst);
         }
       }
-      delete(dst, false);
+      return true;
     } else {
       final Path parent = dst.getParent();
       final FileStatus parentStatus = getFileStatus(parent);
       if (parentStatus.isFile()) {
-        throw new ParentNotDirectoryException("Rename destination parent "
-            + parent + " is a file.");
+        throw new ParentNotDirectoryException("Rename destination parent " + parent + " is a file.");
       }
     }
-    renameInternal(src, dst);
+    return false;
   }
 
   /**

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

@@ -1255,4 +1255,10 @@ private Constants() {
    */
   public static final String PREFETCH_BLOCK_COUNT_KEY = "fs.s3a.prefetch.block.count";
   public static final int PREFETCH_BLOCK_DEFAULT_COUNT = 8;
+
+  /**
+   * Create a fake parent directory on deletes
+   */
+  public static final String CREATE_FAKE_PARENT_DIRECTORY = "fs.s3a.create.fake.parent.directory";
+  public static final boolean CREATE_FAKE_PARENT_DIRECTORY_DEFAULT = true;
 }

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

@@ -22,7 +22,15 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.DelegateToFileSystem;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.apache.hadoop.fs.ParentNotDirectoryException;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.UnresolvedLinkException;
+import org.apache.hadoop.security.AccessControlException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -35,6 +43,8 @@
 @InterfaceStability.Evolving
 public class S3A extends DelegateToFileSystem {
 
+  private final Logger LOG = LoggerFactory.getLogger(this.getClass().getName());
+
   public S3A(URI theUri, Configuration conf)
       throws IOException, URISyntaxException {
     super(theUri, new S3AFileSystem(), conf, "s3a", false);
@@ -55,6 +65,13 @@ public String toString() {
     return sb.toString();
   }
 
+  @Override
+  public void renameInternal(final Path src, final Path dst,
+      boolean overwrite) throws IOException {
+    checkIfCanRename(src, dst, overwrite);
+    fsImpl.rename(src, dst);
+  }
+
   /**
    * Close the file system; the FileContext API doesn't have an explicit close.
    */

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

@@ -420,6 +420,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
    */
   private final Set<Path> deleteOnExit = new TreeSet<>();
 
+  /**
+   * Fake parent directory for cleanup
+   */
+  private boolean createFakeParentDirectory;
+
   /** Add any deprecated keys. */
   @SuppressWarnings("deprecation")
   private static void addDeprecatedKeys() {
@@ -518,6 +523,7 @@ public void initialize(URI name, Configuration originalConf)
           MULTIPART_SIZE, DEFAULT_MULTIPART_SIZE);
       multiPartThreshold = getMultipartSizeProperty(conf,
           MIN_MULTIPART_THRESHOLD, DEFAULT_MIN_MULTIPART_THRESHOLD);
+      createFakeParentDirectory = conf.getBoolean(CREATE_FAKE_PARENT_DIRECTORY, CREATE_FAKE_PARENT_DIRECTORY_DEFAULT);
 
       //check but do not store the block size
       longBytesOption(conf, FS_S3A_BLOCK_SIZE, DEFAULT_BLOCKSIZE, 1);
@@ -2094,7 +2100,7 @@ private Pair<S3AFileStatus, S3AFileStatus> initiateRename(
         // at this point the destination is an empty directory
       } else {
         // source is a file. The destination must be a directory,
-        // empty or not
+        // empty or not existing
         if (dstStatus.isFile()) {
           throw new FileAlreadyExistsException(
               "Failed to rename " + src + " to " + dst
@@ -2156,22 +2162,28 @@ private long innerRename(Path source, Path dest)
     Path src = qualify(source);
     Path dst = qualify(dest);
 
-    LOG.debug("Rename path {} to {}", src, dst);
+    LOG.warn("Rename path {} to {}", src, dst);
 
     String srcKey = pathToKey(src);
     String dstKey = pathToKey(dst);
-
-    Pair<S3AFileStatus, S3AFileStatus> p = initiateRename(src, dst);
+    Pair<S3AFileStatus, S3AFileStatus> p;
+    try{
+       p = initiateRename(src, dst);
+      RenameOperation renameOperation = new RenameOperation(
+          createStoreContext(),
+          src, srcKey, p.getLeft(),
+          dst, dstKey, p.getRight(),
+          new OperationCallbacksImpl(),
+          pageSize);
+      LOG.warn("Please do your thing:");
+      return renameOperation.execute();
+    } catch (Exception e) {
+      LOG.error("This is why it doesn't work", e);
+    }
 
     // Initiate the rename.
     // this will call back into this class via the rename callbacks
-    RenameOperation renameOperation = new RenameOperation(
-        createStoreContext(),
-        src, srcKey, p.getLeft(),
-        dst, dstKey, p.getRight(),
-        new OperationCallbacksImpl(),
-        pageSize);
-    return renameOperation.execute();
+    return 0;
   }
 
   @Override public Token<? extends TokenIdentifier> getFsDelegationToken()
@@ -3253,7 +3265,7 @@ private void createFakeDirectoryIfNecessary(Path f)
   protected void maybeCreateFakeParentDirectory(Path path)
       throws IOException, AmazonClientException {
     Path parent = path.getParent();
-    if (parent != null && !parent.isRoot() && !isUnderMagicCommitPath(parent)) {
+    if (parent != null && !parent.isRoot() && !isUnderMagicCommitPath(parent) && createFakeParentDirectory) {
       createFakeDirectoryIfNecessary(parent);
     }
   }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEmptyDirectory.java

@@ -38,6 +38,7 @@ public void testDirectoryBecomesEmpty() throws Exception {
 
     // 1. set up non-empty dir
     Path dir = path("testEmptyDir");
+    // add a rm of dir to guarantee its not there
     Path child = path("testEmptyDir/dir2");
     mkdirs(child);
 
@@ -47,7 +48,7 @@ public void testDirectoryBecomesEmpty() throws Exception {
     // 2. Make testEmptyDir empty
     assertDeleted(child, false);
     status = getS3AFileStatus(fs, dir);
-
+    // all parent directories are being created and if not then we expect this to fail
     assertEmptyDirectory(true, status);
   }