HDFS-13934 Multipart Uploader APIa

This is ready to g in pending review and the final builder spec.

Specification:

* update the docs to all but the builder
* builder has interface/impl split; extends existing builder implementation
* FileContext support
* add an optional abortUploadsUnderPath(); invaluable for S3A.

Big rework of S3A implementation

* has new S3AMultipartUploaderStatistics interface (implementation from Instrumentation)
* factor WriteOperations interface out of WriteOperationsHelper and use that
* Found and fixed DDB s3guard bug which would fail with >1 file write to the same test. Doesn't show up in existing workflows, but you can upload two files to the same test, and then it surfaces.

Tests

* more!

Change-Id: I1b4aefef51141a53d755e96628d68cebf285f06c

Author: steveloughran

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

@@ -79,11 +79,11 @@ public abstract class AbstractFileSystem implements PathCapabilities {
 
   /** Recording statistics per a file system class. */
   private static final Map<URI, Statistics> 
-      STATISTICS_TABLE = new HashMap<URI, Statistics>();
+      STATISTICS_TABLE = new HashMap<>();
 
   /** Cache of constructors for each file system class. */
   private static final Map<Class<?>, Constructor<?>> CONSTRUCTOR_CACHE = 
-    new ConcurrentHashMap<Class<?>, Constructor<?>>();
+    new ConcurrentHashMap<>();
 
   private static final Class<?>[] URI_CONFIG_ARGS = 
     new Class[]{URI.class, Configuration.class};
@@ -1383,4 +1383,34 @@ public boolean hasPathCapability(final Path path,
       return false;
     }
   }
+
+  /**
+   * Create a multipart uploader.
+   * @param basePath file path under which all files are uploaded
+   * @return a MultipartUploaderBuilder object to build the uploader
+   * @throws IOException if some early checks cause IO failures.
+   * @throws UnsupportedOperationException if support is checked early.
+   */
+  @InterfaceStability.Unstable
+  public MultipartUploaderBuilder createMultipartUploader(Path basePath)
+      throws IOException {
+    methodNotSupported();
+    return null;
+  }
+
+  /**
+   * Helper method that throws an {@link UnsupportedOperationException} for the
+   * current {@link FileSystem} method being called.
+   */
+  protected void methodNotSupported() {
+    // The order of the stacktrace elements is (from top to bottom):
+    //   - java.lang.Thread.getStackTrace
+    //   - org.apache.hadoop.fs.FileSystem.methodNotSupported
+    //   - <the FileSystem method>
+    // therefore, to find out the current method name, we use the element at
+    // index 2.
+    String name = Thread.currentThread().getStackTrace()[2].getMethodName();
+    throw new UnsupportedOperationException(getClass().getCanonicalName() +
+        " does not support method " + name);
+  }
 }

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

@@ -2957,4 +2957,30 @@ public boolean hasPathCapability(Path path, String capability)
         (fs, p) -> fs.hasPathCapability(p, capability));
   }
 
+  /**
+   * Return a set of server default configuration values based on path.
+   * @param path path to fetch server defaults
+   * @return server default configuration values for path
+   * @throws IOException an I/O error occurred
+   */
+  public FsServerDefaults getServerDefaults(final Path path) throws IOException {
+    return FsLinkResolution.resolve(this,
+        fixRelativePart(path),
+        (fs, p) -> fs.getServerDefaults(p));
+  }
+
+  /**
+   * Create a multipart uploader.
+   * @param basePath file path under which all files are uploaded
+   * @return a MultipartUploaderBuilder object to build the uploader
+   * @throws IOException if some early checks cause IO failures.
+   * @throws UnsupportedOperationException if support is checked early.
+   */
+  @InterfaceStability.Unstable
+  public MultipartUploaderBuilder createMultipartUploader(Path basePath)
+      throws IOException {
+    return FsLinkResolution.resolve(this,
+        fixRelativePart(basePath),
+        (fs, p) -> fs.createMultipartUploader(p));
+  }
 }

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

@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -15,6 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.hadoop.fs;
 
 import java.io.Closeable;
@@ -23,23 +24,12 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 
-import static com.google.common.base.Preconditions.checkArgument;
-
 /**
  * MultipartUploader is an interface for copying files multipart and across
- * multiple nodes. Users should:
- * <ol>
- *   <li>Initialize an upload.</li>
- *   <li>Upload parts in any order.</li>
- *   <li>Complete the upload in order to have it materialize in the destination
- *   FS.</li>
- * </ol>
+ * multiple nodes.
  */
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
@@ -52,49 +42,63 @@ public interface MultipartUploader extends Closeable {
    * @return unique identifier associating part uploads.
    * @throws IOException IO failure
    */
-  CompletableFuture<UploadHandle> initialize(Path filePath)
+  CompletableFuture<UploadHandle> startUpload(Path filePath)
       throws IOException;
 
   /**
    * Put part as part of a multipart upload.
    * It is possible to have parts uploaded in any order (or in parallel).
-   * @param filePath Target path for upload (same as {@link #initialize(Path)}).
+   * @param uploadId Identifier from {@link #startUpload(Path)}.
+   * @param partNumber Index of the part relative to others.
+   * @param filePath Target path for upload (same as {@link #startUpload(Path)}).
    * @param inputStream Data for this part. Implementations MUST close this
    * stream after reading in the data.
-   * @param partNumber Index of the part relative to others.
-   * @param uploadId Identifier from {@link #initialize(Path)}.
    * @param lengthInBytes Target length to read from the stream.
    * @return unique PartHandle identifier for the uploaded part.
    * @throws IOException IO failure
    */
-  CompletableFuture<PartHandle> putPart(Path filePath,
+  CompletableFuture<PartHandle> putPart(
+      UploadHandle uploadId,
+      int partNumber,
+      Path filePath,
       InputStream inputStream,
-      int partNumber, UploadHandle uploadId, long lengthInBytes)
+      long lengthInBytes)
       throws IOException;
 
   /**
    * Complete a multipart upload.
-   * @param filePath Target path for upload (same as {@link #initialize(Path)}.
+   * @param uploadId Identifier from {@link #startUpload(Path)}.
+   * @param filePath Target path for upload (same as {@link #startUpload(Path)}.
    * @param handles non-empty map of part number to part handle.
-   *          from {@link #putPart(Path, InputStream, int, UploadHandle, long)}.
-   * @param multipartUploadId Identifier from {@link #initialize(Path)}.
+   *          from {@link #putPart(UploadHandle, int, Path, InputStream, long)}.
    * @return unique PathHandle identifier for the uploaded file.
    * @throws IOException IO failure
    */
-  CompletableFuture<PathHandle> complete(Path filePath,
-      Map<Integer, PartHandle> handles,
-      UploadHandle multipartUploadId)
+  CompletableFuture<PathHandle> complete(
+      UploadHandle uploadId,
+      Path filePath,
+      Map<Integer, PartHandle> handles)
       throws IOException;
 
   /**
    * Aborts a multipart upload.
-   * @param filePath Target path for upload (same as {@link #initialize(Path)}.
-   * @param multipartUploadId Identifier from {@link #initialize(Path)}.
+   * @param uploadId Identifier from {@link #startUpload(Path)}.
+   * @param filePath Target path for upload (same as {@link #startUpload(Path)}.
    * @throws IOException IO failure
-   * @return
+   * @return a future; the operation will have completed
    */
-  CompletableFuture<Void> abort(Path filePath,
-      UploadHandle multipartUploadId)
+  CompletableFuture<Void> abort(UploadHandle uploadId, Path filePath)
       throws IOException;
 
+  /**
+   * Best effort attempt to aborts multipart uploads under a path.
+   * Not all implementations support this, and those which do may
+   * be vulnerable to eventually consistent listings of current uploads
+   * -some may be missed.
+   * @param path path to abort uploads under.
+   * @return a future of the number of entries found; 1 if aborting is unsupported.
+   * @throws IOException IO failure
+   */
+  CompletableFuture<Integer> abortUploadsUnderPath(Path path) throws IOException;
+
 }