HDFS-13934 final tuning of multipart API

especially patch qualification.

I had to add a new operation to the S3A StoreContext class;
took the opportunity to move to a builder API.

Why so -even though I'd argue before we didn't need it?

Well, because S3A DelegationTokens plugin point now takes it, so must their tests.
also tagged as @limitedPrivate for the same reason.

Thinking maybe I should separate that change -and avoid having a guava class in the fields
...won't work with shading, will it?

Change-Id: I19e8f5219d6fe51002c97694af8001453d8eb6b0

Author: steveloughran

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

@@ -131,7 +131,7 @@ protected void checkPutArguments(Path filePath,
    * 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.
+   * @return a future of the number of entries found; -1 if aborting is unsupported.
    * @throws IOException IO failure
    */
   public CompletableFuture<Integer> abortUploadsUnderPath(Path path)

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

@@ -179,13 +179,20 @@ public CompletableFuture<PathHandle> complete(
 
     checkPath(filePath);
     return FutureIOSupport.eval(() ->
-        innerComplete(filePath, handleMap, uploadId));
+        innerComplete(uploadId, filePath, handleMap));
   }
 
-  public PathHandle innerComplete(
-      Path filePath,
-      Map<Integer, PartHandle> handleMap,
-      UploadHandle multipartUploadId) throws IOException {
+  /**
+   * The upload complete operation.
+   * @param multipartUploadId the ID of the upload
+   * @param filePath path
+   * @param handleMap map of handles
+   * @return the path handle
+   * @throws IOException failure
+   */
+  private PathHandle innerComplete(
+      UploadHandle multipartUploadId, Path filePath,
+      Map<Integer, PartHandle> handleMap) throws IOException {
 
     checkPath(filePath);
 

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

@@ -91,7 +91,7 @@ protected MultipartUploaderBuilderImpl(@Nonnull FileContext fc,
    */
   protected MultipartUploaderBuilderImpl(@Nonnull FileSystem fileSystem,
       @Nonnull Path p) {
-    super(checkNotNull(p));
+    super(fileSystem.makeQualified(checkNotNull(p)));
     checkNotNull(fileSystem);
     fs = fileSystem;
     bufferSize = fs.getConf().getInt(IO_FILE_BUFFER_SIZE_KEY,

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

@@ -112,6 +112,7 @@
 import org.apache.hadoop.fs.s3a.impl.S3AMultipartUploaderBuilder;
 import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
+import org.apache.hadoop.fs.s3a.impl.StoreContextBuilder;
 import org.apache.hadoop.fs.s3a.s3guard.BulkOperationState;
 import org.apache.hadoop.fs.s3a.select.InternalSelectConstants;
 import org.apache.hadoop.io.IOUtils;
@@ -4726,24 +4727,24 @@ public MultipartUploaderBuilder createMultipartUploader(final Path basePath)
    */
   @InterfaceAudience.Private
   public StoreContext createStoreContext() {
-    return new StoreContext(
-        getUri(),
-        getBucket(),
-        getConf(),
-        getUsername(),
-        owner,
-        boundedThreadPool,
-        executorCapacity,
-        invoker,
-        getInstrumentation(),
-        getStorageStatistics(),
-        getInputPolicy(),
-        changeDetectionPolicy,
-        enableMultiObjectsDelete,
-        metadataStore,
-        useListV1,
-        new ContextAccessorsImpl(),
-        getTtlTimeProvider());
+    return new StoreContextBuilder().setFsURI(getUri())
+        .setBucket(getBucket())
+        .setConfiguration(getConf())
+        .setUsername(getUsername())
+        .setOwner(owner)
+        .setExecutor(boundedThreadPool)
+        .setExecutorCapacity(executorCapacity)
+        .setInvoker(invoker)
+        .setInstrumentation(getInstrumentation())
+        .setStorageStatistics(getStorageStatistics())
+        .setInputPolicy(getInputPolicy())
+        .setChangeDetectionPolicy(changeDetectionPolicy)
+        .setMultiObjectDeleteEnabled(enableMultiObjectsDelete)
+        .setMetadataStore(metadataStore)
+        .setUseListV1(useListV1)
+        .setContextAccessors(new ContextAccessorsImpl())
+        .setTimeProvider(getTtlTimeProvider())
+        .build();
   }
 
   /**
@@ -4771,5 +4772,10 @@ public File createTempFile(final String prefix, final long size)
     public String getBucketLocation() throws IOException {
       return S3AFileSystem.this.getBucketLocation();
     }
+
+    @Override
+    public Path makeQualified(final Path path) {
+      return S3AFileSystem.this.makeQualified(path);
+    }
   }
 }

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

@@ -73,4 +73,12 @@ public interface ContextAccessors {
    */
   @Retries.RetryTranslated
   String getBucketLocation() throws IOException;
+
+  /**
+   * Qualify a path.
+   *
+   * @param path path to qualify/normalize
+   * @return possibly new path.
+   */
+  Path makeQualified(Path path);
 }

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

@@ -98,7 +98,7 @@ class S3AMultipartUploader extends AbstractMultipartUploader {
       final WriteOperations writeOperations,
       final StoreContext context,
       final S3AMultipartUploaderStatistics statistics) {
-    super(builder.getPath());
+    super(context.makeQualified(builder.getPath()));
     this.builder = builder;
     this.writeOperations = writeOperations;
     this.context = context;
@@ -132,8 +132,9 @@ private synchronized BulkOperationState retrieveOperationState()
   @Override
   public CompletableFuture<UploadHandle> startUpload(Path filePath)
       throws IOException {
-    checkPath(filePath);
-    String key = context.pathToKey(filePath);
+    Path dest = context.makeQualified(filePath);
+    checkPath(dest);
+    String key = context.pathToKey(dest);
     return context.submit(new CompletableFuture<>(),
         () -> {
           String uploadId = writeOperations.initiateMultiPartUpload(key);
@@ -150,11 +151,12 @@ public CompletableFuture<PartHandle> putPart(UploadHandle uploadId,
       InputStream inputStream,
       long lengthInBytes)
       throws IOException {
-    checkPutArguments(filePath, inputStream, partNumber, uploadId,
+    Path dest = context.makeQualified(filePath);
+    checkPutArguments(dest, inputStream, partNumber, uploadId,
         lengthInBytes);
     byte[] uploadIdBytes = uploadId.toByteArray();
     checkUploadId(uploadIdBytes);
-    String key = context.pathToKey(filePath);
+    String key = context.pathToKey(dest);
     String uploadIdString = new String(uploadIdBytes, 0, uploadIdBytes.length,
         Charsets.UTF_8);
     return context.submit(new CompletableFuture<>(),
@@ -180,15 +182,16 @@ public CompletableFuture<PathHandle> complete(
       Path filePath,
       Map<Integer, PartHandle> handleMap)
       throws IOException {
-    checkPath(filePath);
+    Path dest = context.makeQualified(filePath);
+    checkPath(dest);
     byte[] uploadIdBytes = uploadHandle.toByteArray();
     checkUploadId(uploadIdBytes);
     checkPartHandles(handleMap);
     List<Map.Entry<Integer, PartHandle>> handles =
         new ArrayList<>(handleMap.entrySet());
     handles.sort(Comparator.comparingInt(Map.Entry::getKey));
     int count = handles.size();
-    String key = context.pathToKey(filePath);
+    String key = context.pathToKey(dest);
 
     String uploadIdStr = new String(uploadIdBytes, 0, uploadIdBytes.length,
         Charsets.UTF_8);
@@ -233,15 +236,16 @@ public CompletableFuture<Void> abort(
       UploadHandle uploadId,
       Path filePath)
       throws IOException {
-    checkPath(filePath);
+    Path dest = context.makeQualified(filePath);
+    checkPath(dest);
     final byte[] uploadIdBytes = uploadId.toByteArray();
     checkUploadId(uploadIdBytes);
     String uploadIdString = new String(uploadIdBytes, 0, uploadIdBytes.length,
         Charsets.UTF_8);
     return context.submit(new CompletableFuture<>(),
         () -> {
           writeOperations.abortMultipartCommit(
-              context.pathToKey(filePath),
+              context.pathToKey(dest),
               uploadIdString);
           statistics.uploadAborted();
           return null;

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

@@ -62,4 +62,5 @@ public S3AMultipartUploader build()
     return new S3AMultipartUploader(this, writeOperations, context, statistics);
   }
 
+
 }

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

@@ -52,9 +52,10 @@
  * their own.
  *
  * <i>Warning:</i> this really is private and unstable. Do not use
- * outside the org.apache.hadoop.fs.s3a package.
+ * outside the org.apache.hadoop.fs.s3a package, or in extension points
+ * such as DelegationTokens.
  */
-@InterfaceAudience.Private
+@InterfaceAudience.LimitedPrivate("S3A Filesystem and extensions")
 @InterfaceStability.Unstable
 public class StoreContext {
 
@@ -117,8 +118,7 @@ public class StoreContext {
 
   /**
    * Instantiate.
-   * No attempt to use a builder here as outside tests
-   * this should only be created in the S3AFileSystem.
+   * @deprecated as public method: use {@link StoreContextBuilder}.
    */
   public StoreContext(
       final URI fsURI,
@@ -229,6 +229,16 @@ public String pathToKey(Path path) {
     return contextAccessors.pathToKey(path);
   }
 
+  /**
+   * Qualify a path.
+   *
+   * @param path path to qualify/normalize
+   * @return possibly new path.
+   */
+  public Path makeQualified(Path path) {
+    return contextAccessors.makeQualified(path);
+  }
+
   /**
    * Get the storage statistics of this filesystem.
    * @return the storage statistics