HADOOP-16848. Refactoring: initial layering

First PoC of My planned layout model of the S3A FS.

* There's a raw layer and a guarded layer
* which are instantiated in sequence in a separate executor from S3AFileSystem.initalize

And whose accessors block until completed or rethrow failures.

The layers are being handed in all their dependencies from FS.initialize()
and we currently block until started.

What I plan to do, in a future iteration, is:

* each layer extracts their own settings from the config and stores locally
  (list version, upload size etc)
* have each layer instantiate their internal classes (AWS S3 client, transfer manager) internally
* Also async create: metastore, DT binding
* And all startup actions (check bucket, init multipart, ...)

Then
* move ops to the layers, raw* -> rawStore; inner -> S3AStore
* move WriteOperationHelper, SelectBinding, etc, to all work against S3AStore rather than FS.

S3AStore will become where most of the code moves to; S3AFilesystem more of the init and binding to hadoop FS API.
RawS3A will be the accessor through which all AWS client access goes.

Not going to change: all accessors on S3AFileSystem...not just tests use it but some external code (cloudstore) needs it to get at low level S3A, etc.

Change-Id: I998c0d61cce2ee7fd0be804bf21da6b68fd69a6f

HADOOP-16583 refactoring RequestFactory and RawS3A

This moves most of the s3 client interaction into RawS3AImpl,
Mainly just by moving the methods from S3AFileSystem.

One key finding was that we can put all the code to create
Request classes for the AWS SDK into its own factory,
So ensure everything is set up consistently and keeping what
is mainly housekeeping out of the way of everything else.

We can do that rework immediately, as it doesn't require
the rest of the layering.

Also, the rawS3A() accessor no longer raises IOEs, it was
Used into many places where that wasn't allowed.
I sort of expected that.
There's an awaitQuietly() (todo: change name) method
which returns all failures as RTEs rather than IOEs.

For the S3A FS API entry points, we still need to raise
the normal startup exceptions, so we will need extra block/validate there
after which, maybe, we can change the logic in these new getters just to
raise an exception if their refs are null, rather than block.

Change-Id: I3d15a46dd3034fc5d34dbaf01aaddba462d63a9d

HADOOP-16583. refactoring -fix regression

I'd pulled an import which wasn't used, but a recent change reinstated its
need

Change-Id: I12b0725a14a6e7bae1ecf93fcbd4d05e1eacb03e

HADOOP-16848. Building against trunk

Change-Id: Id1af0763998dd285bb2476298cdb94edf809b67c

Author: steveloughran

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

@@ -55,7 +55,7 @@ private MultipartUtils() { }
    *               bucket will be searched.
    * @return an iterator of matching uploads
    */
-  static MultipartUtils.UploadIterator listMultipartUploads(AmazonS3 s3,
+  public static MultipartUtils.UploadIterator listMultipartUploads(AmazonS3 s3,
       Invoker invoker, String bucketName, int maxKeys, @Nullable String prefix)
       throws IOException {
     return new MultipartUtils.UploadIterator(s3, invoker, bucketName, maxKeys,

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

@@ -51,6 +51,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathIOException;
+import org.apache.hadoop.fs.s3a.impl.StoreContext;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.s3a.s3guard.BulkOperationState;
 import org.apache.hadoop.fs.s3a.s3guard.S3Guard;
@@ -116,6 +117,8 @@ public class WriteOperationHelper implements WriteOperations {
    */
   private final S3AStatisticsContext statisticsContext;
 
+  private final StoreContext context;
+
   /**
    * Constructor.
    * @param owner owner FS creating the helper
@@ -131,7 +134,8 @@ protected WriteOperationHelper(S3AFileSystem owner,
         this::operationRetried);
     this.conf = conf;
     this.statisticsContext = statisticsContext;
-    bucket = owner.getBucket();
+    this.context = owner.createStoreContext();
+    this.bucket = owner.getBucket();
   }
 
   /**
@@ -241,11 +245,8 @@ public ObjectMetadata newObjectMetadata(long length) {
   public String initiateMultiPartUpload(String destKey) throws IOException {
     LOG.debug("Initiating Multipart upload to {}", destKey);
     final InitiateMultipartUploadRequest initiateMPURequest =
-        new InitiateMultipartUploadRequest(bucket,
-            destKey,
-            newObjectMetadata(-1));
-    initiateMPURequest.setCannedACL(owner.getCannedACL());
-    owner.setOptionalMultipartUploadRequestParameters(initiateMPURequest);
+        context.getRequestFactory().newMultipartUploadRequest(
+            destKey);
 
     return retry("initiate MultiPartUpload", destKey, true,
         () -> owner.initiateMultipartUpload(initiateMPURequest).getUploadId());
@@ -432,50 +433,8 @@ public UploadPartRequest newUploadPartRequest(
       File sourceFile,
       Long offset) throws PathIOException {
     checkNotNull(uploadId);
-    // exactly one source must be set; xor verifies this
-    checkArgument((uploadStream != null) ^ (sourceFile != null),
-        "Data source");
-    checkArgument(size >= 0, "Invalid partition size %s", size);
-    checkArgument(partNumber > 0,
-        "partNumber must be between 1 and %s inclusive, but is %s",
-            DEFAULT_UPLOAD_PART_COUNT_LIMIT, partNumber);
-
-    LOG.debug("Creating part upload request for {} #{} size {}",
-        uploadId, partNumber, size);
-    long partCountLimit = longOption(conf,
-        UPLOAD_PART_COUNT_LIMIT,
-        DEFAULT_UPLOAD_PART_COUNT_LIMIT,
-        1);
-    if (partCountLimit != DEFAULT_UPLOAD_PART_COUNT_LIMIT) {
-      LOG.warn("Configuration property {} shouldn't be overridden by client",
-              UPLOAD_PART_COUNT_LIMIT);
-    }
-    final String pathErrorMsg = "Number of parts in multipart upload exceeded."
-        + " Current part count = %s, Part count limit = %s ";
-    if (partNumber > partCountLimit) {
-      throw new PathIOException(destKey,
-          String.format(pathErrorMsg, partNumber, partCountLimit));
-    }
-    UploadPartRequest request = new UploadPartRequest()
-        .withBucketName(bucket)
-        .withKey(destKey)
-        .withUploadId(uploadId)
-        .withPartNumber(partNumber)
-        .withPartSize(size);
-    if (uploadStream != null) {
-      // there's an upload stream. Bind to it.
-      request.setInputStream(uploadStream);
-    } else {
-      checkArgument(sourceFile.exists(),
-          "Source file does not exist: %s", sourceFile);
-      checkArgument(offset >= 0, "Invalid offset %s", offset);
-      long length = sourceFile.length();
-      checkArgument(offset == 0 || offset < length,
-          "Offset %s beyond length of file %s", offset, length);
-      request.setFile(sourceFile);
-      request.setFileOffset(offset);
-    }
-    return request;
+
+    return context.getRequestFactory().newUploadPartRequest(destKey, uploadId, partNumber, size, uploadStream, sourceFile, offset);
   }
 
   /**
@@ -631,10 +590,8 @@ public Configuration getConf() {
    * @return the request
    */
   public SelectObjectContentRequest newSelectRequest(Path path) {
-    SelectObjectContentRequest request = new SelectObjectContentRequest();
-    request.setBucketName(bucket);
-    request.setKey(owner.pathToKey(path));
-    return request;
+    return context.getRequestFactory().newSelectRequest(
+        context.pathToKey(path));
   }
 
   /**

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

@@ -24,7 +24,6 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.apache.hadoop.fs.s3a.Statistic;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTracker;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
@@ -48,7 +47,7 @@
 public class MagicCommitIntegration {
   private static final Logger LOG =
       LoggerFactory.getLogger(MagicCommitIntegration.class);
-  private final S3AFileSystem owner;
+  private final StoreContext owner;
   private final boolean magicCommitEnabled;
 
   private final StoreContext storeContext;
@@ -58,7 +57,7 @@ public class MagicCommitIntegration {
    * @param owner owner class
    * @param magicCommitEnabled is magic commit enabled.
    */
-  public MagicCommitIntegration(S3AFileSystem owner,
+  public MagicCommitIntegration(StoreContext owner,
       boolean magicCommitEnabled) {
     this.owner = owner;
     this.magicCommitEnabled = magicCommitEnabled;

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

@@ -0,0 +1,99 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.impl;
+
+import java.util.Objects;
+
+import org.apache.hadoop.fs.s3a.Invoker;
+import org.apache.hadoop.fs.s3a.Statistic;
+import org.apache.hadoop.service.AbstractService;
+
+public abstract class AbstractS3AService
+    extends AbstractService
+    implements S3AService {
+
+  private StoreContext storeContext;
+
+  protected AbstractS3AService(final String name) {
+    super(name);
+  }
+
+  protected void bind(final StoreContext storeContext) {
+    this.storeContext = storeContext;
+  }
+
+  @Override
+  public StoreContext getStoreContext() {
+    return storeContext;
+  }
+
+  /**
+   * Validate the state of the service, then start the service.
+   * Service start may be async.
+   * @throws Exception if initialization failed.
+   */
+  @Override
+  protected void serviceStart() throws Exception {
+    Objects.requireNonNull(storeContext, () ->
+        "not initialized with store context: " + getName());
+  }
+
+  /**
+   * Increment a statistic by 1.
+   * This increments both the instrumentation and storage statistics.
+   * @param statistic The operation to increment
+   */
+  protected void incrementStatistic(Statistic statistic) {
+    incrementStatistic(statistic, 1);
+  }
+
+  /**
+   * Increment a statistic by a specific value.
+   * This increments both the instrumentation and storage statistics.
+   * @param statistic The operation to increment
+   * @param count the count to increment
+   */
+  protected void incrementStatistic(Statistic statistic, long count) {
+    // todo
+  }
+
+  /**
+   * Increment read operations.
+   */
+  public void incrementReadOperations() {
+    // statistics.incrementReadOps(1);
+  }
+
+  /**
+   * Increment the write operation counter.
+   * This is somewhat inaccurate, as it appears to be invoked more
+   * often than needed in progress callbacks.
+   */
+  public void incrementWriteOperations() {
+//    statistics.incrementWriteOps(1);
+  }
+
+  protected String getBucket() {
+    return getStoreContext().getBucket();
+  }
+
+  protected Invoker getInvoker() {
+    return getStoreContext().getInvoker();
+  }
+}

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

@@ -21,6 +21,7 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.io.InterruptedIOException;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
@@ -118,7 +119,8 @@ public static <T> void waitForCompletion(
             new DurationInfo(LOG, false, "Waiting for task completion")) {
       future.join();
     } catch (CancellationException e) {
-      throw new IOException(e);
+      throw (InterruptedIOException)
+          new InterruptedIOException(e.toString()).initCause(e);
     } catch (CompletionException e) {
       raiseInnerCause(e);
     }

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

@@ -26,6 +26,7 @@
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.Retries;
+import org.apache.hadoop.fs.s3a.WriteOperationHelper;
 
 /**
  * An interface to implement for providing accessors to
@@ -94,4 +95,10 @@ public interface ContextAccessors {
   @Retries.RetryTranslated
   ObjectMetadata getObjectMetadata(String key) throws IOException;
 
+
+  /**
+   * Get a write operation helper.
+   * @return a write operation helper instance.
+   */
+  WriteOperationHelper getWriteOperationHelper();
 }

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

@@ -37,6 +37,16 @@
  */
 public final class InternalConstants {
 
+  /**
+   * This declared delete as idempotent.
+   * This is an "interesting" topic in past Hadoop FS work.
+   * Essentially: with a single caller, DELETE is idempotent
+   * but in a shared filesystem, it is is very much not so.
+   * Here, on the basis that isn't a filesystem with consistency guarantees,
+   * retryable results in files being deleted.
+  */
+  public static final boolean DELETE_CONSIDERED_IDEMPOTENT = true;
+
   private InternalConstants() {
   }