HADOOP-16877. S3A FS deleteOnExit to skip the exists check.

* new override method S3AFileSystem.deleteOnExit() which skips the exists
  check.
* FileSystem.processDeleteOnExit() skips exists checks too; relies on delete()
  to do its work.
* make sure all the delete/cancel/process deleteOnExit operations consistently
  qualify paths, especially the list of paths to delete

Change-Id: Icd99615f7e6adbb7a4c6e1cfdff44edc21956961

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

@@ -1664,20 +1664,29 @@ public abstract class FileSystem extends Configured
    *   <li>Connectivity problems with a remote filesystem may delay shutdown
    *   further, and may cause the files to not be deleted.</li>
    * </ol>
-   * @param f the path to delete.
+   * @param f the path to delete; this may be unqualified.
    * @return  true if deleteOnExit is successful, otherwise false.
    * @throws IOException IO failure
    */
   public boolean deleteOnExit(Path f) throws IOException {
-    if (!exists(f)) {
+    Path path = makeQualified(f);
+    if (!exists(path)) {
       return false;
     }
-    synchronized (deleteOnExit) {
+    queueForDeletion(path);
-      deleteOnExit.add(f);
-    }
     return true;
   }
 
+  /**
+   * Add a path to the set of paths to delete on shutdown.
+   * @param path fully qualified path.
+   */
+  protected void queueForDeletion(final Path path) {
+    synchronized (deleteOnExit) {
+      deleteOnExit.add(path);
+    }
+  }
+
   /**
    * Cancel the scheduled deletion of the path when the FileSystem is closed.
    * @param f the path to cancel deletion
@@ -1685,7 +1694,7 @@ public abstract class FileSystem extends Configured
    */
   public boolean cancelDeleteOnExit(Path f) {
     synchronized (deleteOnExit) {
-      return deleteOnExit.remove(f);
+      return deleteOnExit.remove(makeQualified(f));
     }
   }
 
@@ -1702,10 +1711,8 @@ public abstract class FileSystem extends Configured
       for (Iterator<Path> iter = deleteOnExit.iterator(); iter.hasNext();) {
         Path path = iter.next();
         try {
-          if (exists(path)) {
           delete(path, true);
         }
-        }
         catch (IOException e) {
           LOGGER.info("Ignoring failure to deleteOnExit for path {}", path);
         }

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

@@ -2484,6 +2484,31 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
     }
   }
 
+  /**
+   * This override queues the path for deletion without
+   * checking for existence. This:
+   * <ol>
+   *   <li>
+   *     Saves HTTP requests.
+   *   </li>
+   *   <li>
+   *     Queues files before they have been materialized through
+   *     S3ABlockOutputStream.close() or similar.
+   *   </li>
+   *   <li>
+   *     Does not add 404 entries to the S3 load balancers.
+   *   </li>
+   * </ol>
+   * @param f the path to delete; this may be unqualified.
+   * @return true, always.
+   * @throws IOException
+   */
+  @Override
+  public boolean deleteOnExit(Path f) throws IOException {
+    queueForDeletion(makeQualified(f));
+    return true;
+  }
+
   /**
    * Create a fake directory if required.
    * That is: it is not the root path and the path does not exist.