HDFS-13934. Multipart uploaders to be created through FileSystem/FileContext.

This is my proposal for an API for multipart uploads through the FS, rather than via service loading.

It builds on the work done in createFile() and openFile() for builders and futures.

* a builder API for configuring the uploader, including specifying the permissions of uploaded files.
* MultipartUploader is now interface; org.apache.hadoop.fs.impl.AbstractMultipartUploader is the base class for implementations to extend.
* all the operations in the interface now return FutureCompletable<>; implementations can perform the operations synchronously or asynchronously.
* there is a new common path capability for the hasPathCapabilities() probe.

FileSystemMultipartUploader and its builder moved to fs.impl; I needed a class org.apache.hadoop.fs.InternalOperations to get at rename/3 from the new package. It really is time to make the good rename method public.

S3AMultipartUploader moved to fs.s3a.impl, alongside its builder. The uploader does not get a reference to the base FS; it gets a StoreContext and WriteOperationHelper and has to work with them. All operations and now async and executed in the executor offered by the StoreContext.

Tests are all fixed up to cope with the changed API.

The builders can be created from the file system; we can also extend file context and the filter classes... Though I would not pass the operation through checksum FS as the uploads will not be checksummed.

One thing I am unsure about is viewFS: even though we can do with the remapping when the builder is created, the actual uploads need the full path in the store. That is unless, this and the WiP copy operation both take a path remapper in the builder which is then used to map from View FS paths to ones in their store. Suggestions here are welcome.

No changes into the specification; yet. Let's get feedback first.

Change-Id: Ib526fe6db8e9282c634181be231d4124ea6c78fc

Author: steveloughran

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

@@ -131,4 +131,12 @@ private CommonPathCapabilities() {
   @InterfaceStability.Unstable
   public static final String FS_EXPERIMENTAL_BATCH_LISTING =
       "fs.capability.batch.listing";
+
+  /**
+   * Does the store support multipart uploading?
+   * Value: {@value}.
+   */
+  public static final String FS_MULTIPART_UPLOADER =
+      "fs.capability.multipart.uploader";
+
 }

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

@@ -4644,4 +4644,17 @@ public CompletableFuture<FSDataInputStream> build() throws IOException {
 
   }
 
+  /**
+   * 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;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSMultipartUploaderFactory.java renamed to hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/InternalOperations.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,26 +15,26 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.hdfs;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileSystemMultipartUploader;
-import org.apache.hadoop.fs.MultipartUploader;
-import org.apache.hadoop.fs.MultipartUploaderFactory;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+package org.apache.hadoop.fs;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
 
 /**
- * Support for HDFS multipart uploads, built on
- * {@link FileSystem#concat(Path, Path[])}.
+ *   This method allows access to Package-scoped operations from classes
+ *   in org.apache.hadoop.fs.impl and other file system implementations
+ *   in the hadoop modules.
+ *   This is absolutely not for used by any other application or library.
  */
-public class DFSMultipartUploaderFactory extends MultipartUploaderFactory {
-  protected MultipartUploader createMultipartUploader(FileSystem fs,
-      Configuration conf) {
-    if (fs.getScheme().equals(HdfsConstants.HDFS_URI_SCHEME)) {
-      return new FileSystemMultipartUploader(fs);
-    }
-    return null;
+@InterfaceAudience.Private
+public class InternalOperations {
+
+  @SuppressWarnings("deprecation") // rename w/ OVERWRITE
+  public void rename(FileSystem fs, final Path src, final Path dst,
+    final Options.Rename...options) throws IOException {
+    fs.rename(src, dst, options);
   }
 }

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

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,28 +41,19 @@
  *   FS.</li>
  * </ol>
  */
-@InterfaceAudience.Private
+@InterfaceAudience.Public
 @InterfaceStability.Unstable
-public abstract class MultipartUploader implements Closeable {
-  public static final Logger LOG =
-      LoggerFactory.getLogger(MultipartUploader.class);
+public interface MultipartUploader extends Closeable {
 
-  /**
-   * Perform any cleanup.
-   * The upload is not required to support any operations after this.
-   * @throws IOException problems on close.
-   */
-  @Override
-  public void close() throws IOException {
-  }
 
   /**
    * Initialize a multipart upload.
    * @param filePath Target path for upload.
    * @return unique identifier associating part uploads.
    * @throws IOException IO failure
    */
-  public abstract UploadHandle initialize(Path filePath) throws IOException;
+  CompletableFuture<UploadHandle> initialize(Path filePath)
+      throws IOException;
 
   /**
    * Put part as part of a multipart upload.
@@ -75,7 +67,8 @@ public void close() throws IOException {
    * @return unique PartHandle identifier for the uploaded part.
    * @throws IOException IO failure
    */
-  public abstract PartHandle putPart(Path filePath, InputStream inputStream,
+  CompletableFuture<PartHandle> putPart(Path filePath,
+      InputStream inputStream,
       int partNumber, UploadHandle uploadId, long lengthInBytes)
       throws IOException;
 
@@ -88,7 +81,7 @@ public abstract PartHandle putPart(Path filePath, InputStream inputStream,
    * @return unique PathHandle identifier for the uploaded file.
    * @throws IOException IO failure
    */
-  public abstract PathHandle complete(Path filePath,
+  CompletableFuture<PathHandle> complete(Path filePath,
       Map<Integer, PartHandle> handles,
       UploadHandle multipartUploadId)
       throws IOException;
@@ -98,57 +91,10 @@ public abstract PathHandle complete(Path filePath,
    * @param filePath Target path for upload (same as {@link #initialize(Path)}.
    * @param multipartUploadId Identifier from {@link #initialize(Path)}.
    * @throws IOException IO failure
+   * @return
    */
-  public abstract void abort(Path filePath, UploadHandle multipartUploadId)
+  CompletableFuture<Void> abort(Path filePath,
+      UploadHandle multipartUploadId)
       throws IOException;
 
-  /**
-   * Utility method to validate uploadIDs.
-   * @param uploadId Upload ID
-   * @throws IllegalArgumentException invalid ID
-   */
-  protected void checkUploadId(byte[] uploadId)
-      throws IllegalArgumentException {
-    checkArgument(uploadId != null, "null uploadId");
-    checkArgument(uploadId.length > 0,
-        "Empty UploadId is not valid");
-  }
-
-  /**
-   * Utility method to validate partHandles.
-   * @param partHandles handles
-   * @throws IllegalArgumentException if the parts are invalid
-   */
-  protected void checkPartHandles(Map<Integer, PartHandle> partHandles) {
-    checkArgument(!partHandles.isEmpty(),
-        "Empty upload");
-    partHandles.keySet()
-        .stream()
-        .forEach(key ->
-            checkArgument(key > 0,
-                "Invalid part handle index %s", key));
-  }
-
-  /**
-   * Check all the arguments to the
-   * {@link #putPart(Path, InputStream, int, UploadHandle, long)} operation.
-   * @param filePath Target path for upload (same as {@link #initialize(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.
-   * @throws IllegalArgumentException invalid argument
-   */
-  protected void checkPutArguments(Path filePath,
-      InputStream inputStream,
-      int partNumber,
-      UploadHandle uploadId,
-      long lengthInBytes) throws IllegalArgumentException {
-    checkArgument(filePath != null, "null filePath");
-    checkArgument(inputStream != null, "null inputStream");
-    checkArgument(partNumber > 0, "Invalid part number: %d", partNumber);
-    checkArgument(uploadId != null, "null uploadId");
-    checkArgument(lengthInBytes >= 0, "Invalid part length: %d", lengthInBytes);
-  }
 }