HADOOP-16848. S3A refactoring

RequestFactory adds factory construction and full set of operations.
Moves all S3 IO into the RawS3A class, takes factory
S3 select API call goes into the store
remove attempt at async init. Too complex.

Change-Id: I36bb2884ebe0d7183f99f25860f79e3796a112fb

Author: steveloughran

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

@@ -23,12 +23,10 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
 import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
@@ -51,20 +49,15 @@
 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.RawS3A;
 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;
-import org.apache.hadoop.fs.s3a.select.SelectBinding;
-import org.apache.hadoop.util.DurationInfo;
 import org.apache.hadoop.util.functional.CallableRaisingIOE;
 
-import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.hadoop.fs.s3a.Invoker.*;
-import static org.apache.hadoop.fs.s3a.S3AUtils.longOption;
-import static org.apache.hadoop.fs.s3a.impl.InternalConstants.DEFAULT_UPLOAD_PART_COUNT_LIMIT;
-import static org.apache.hadoop.fs.s3a.impl.InternalConstants.UPLOAD_PART_COUNT_LIMIT;
 
 /**
  * Helper for low-level operations against an S3 Bucket for writing data,
@@ -112,6 +105,9 @@ public class WriteOperationHelper implements WriteOperations {
   /** Bucket of the owner FS. */
   private final String bucket;
 
+  /** Raw S3A implementation to invoke. */
+  private final RawS3A rawS3A;
+
   /**
    * statistics context.
    */
@@ -124,18 +120,20 @@ public class WriteOperationHelper implements WriteOperations {
    * @param owner owner FS creating the helper
    * @param conf Configuration object
    * @param statisticsContext statistics context
-   *
+   * @param rawS3A raw S3A implementation.
    */
   protected WriteOperationHelper(S3AFileSystem owner,
       Configuration conf,
-      S3AStatisticsContext statisticsContext) {
+      S3AStatisticsContext statisticsContext,
+      RawS3A rawS3A) {
     this.owner = owner;
     this.invoker = new Invoker(new S3ARetryPolicy(conf),
         this::operationRetried);
     this.conf = conf;
     this.statisticsContext = statisticsContext;
     this.context = owner.createStoreContext();
     this.bucket = owner.getBucket();
+    this.rawS3A = rawS3A;
   }
 
   /**
@@ -231,7 +229,7 @@ public void writeFailed(Exception ex) {
    * @return a new metadata instance
    */
   public ObjectMetadata newObjectMetadata(long length) {
-    return owner.newObjectMetadata(length);
+    return context.getRequestFactory().newObjectMetadata(length);
   }
 
   /**
@@ -249,7 +247,7 @@ public String initiateMultiPartUpload(String destKey) throws IOException {
             destKey);
 
     return retry("initiate MultiPartUpload", destKey, true,
-        () -> owner.initiateMultipartUpload(initiateMPURequest).getUploadId());
+        () -> rawS3A.initiateMultipartUpload(initiateMPURequest).getUploadId());
   }
 
   /**
@@ -279,18 +277,16 @@ private CompleteMultipartUploadResult finalizeMultipartUpload(
       throw new PathIOException(destKey,
           "No upload parts in multipart upload");
     }
+
+    CompleteMultipartUploadRequest request =
+        context.getRequestFactory().newCompleteMultipartUploadRequest(
+            destKey, uploadId, partETags);
     CompleteMultipartUploadResult uploadResult =
         invoker.retry("Completing multipart upload", destKey,
             true,
             retrying,
             () -> {
-              // a copy of the list is required, so that the AWS SDK doesn't
-              // attempt to sort an unmodifiable list.
-              return owner.getAmazonS3Client().completeMultipartUpload(
-                  new CompleteMultipartUploadRequest(bucket,
-                      destKey,
-                      uploadId,
-                      new ArrayList<>(partETags)));
+              return rawS3A.completeMultipartUpload(request);
             }
     );
     owner.finishedWrite(destKey, length, uploadResult.getETag(),
@@ -432,9 +428,9 @@ public UploadPartRequest newUploadPartRequest(
       InputStream uploadStream,
       File sourceFile,
       Long offset) throws PathIOException {
-    checkNotNull(uploadId);
 
-    return context.getRequestFactory().newUploadPartRequest(destKey, uploadId, partNumber, size, uploadStream, sourceFile, offset);
+    return context.getRequestFactory().newUploadPartRequest(
+        destKey, uploadId, partNumber, size, uploadStream, sourceFile, offset);
   }
 
   /**
@@ -610,33 +606,6 @@ public SelectObjectContentResult select(
       final SelectObjectContentRequest request,
       final String action)
       throws IOException {
-    String bucketName = request.getBucketName();
-    Preconditions.checkArgument(bucket.equals(bucketName),
-        "wrong bucket: %s", bucketName);
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("Initiating select call {} {}",
-          source, request.getExpression());
-      LOG.debug(SelectBinding.toString(request));
-    }
-    return invoker.retry(
-        action,
-        source.toString(),
-        true,
-        () -> {
-          try (DurationInfo ignored =
-                   new DurationInfo(LOG, "S3 Select operation")) {
-            try {
-              return owner.getAmazonS3Client().selectObjectContent(request);
-            } catch (AmazonS3Exception e) {
-              LOG.error("Failure of S3 Select request against {}",
-                  source);
-              LOG.debug("S3 Select request against {}:\n{}",
-                  source,
-                  SelectBinding.toString(request),
-                  e);
-              throw e;
-            }
-          }
-        });
+    return owner.store().select(source, request, action);
   }
 }

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

@@ -25,6 +25,7 @@
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.Statistic;
+import org.apache.hadoop.fs.s3a.WriteOperationHelper;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTracker;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 
@@ -47,21 +48,25 @@
 public class MagicCommitIntegration {
   private static final Logger LOG =
       LoggerFactory.getLogger(MagicCommitIntegration.class);
-  private final StoreContext owner;
   private final boolean magicCommitEnabled;
 
   private final StoreContext storeContext;
 
+  private final WriteOperationHelper writeOperationHelper;
+
   /**
    * Instantiate.
-   * @param owner owner class
+   * @param storeContext store context
+   * @param writeOperationHelper helper
    * @param magicCommitEnabled is magic commit enabled.
    */
-  public MagicCommitIntegration(StoreContext owner,
-      boolean magicCommitEnabled) {
-    this.owner = owner;
+  public MagicCommitIntegration(
+      final StoreContext storeContext,
+      final WriteOperationHelper writeOperationHelper,
+      final boolean magicCommitEnabled) {
     this.magicCommitEnabled = magicCommitEnabled;
-    this.storeContext = owner.createStoreContext();
+    this.storeContext = storeContext;
+    this.writeOperationHelper = writeOperationHelper;
   }
 
   /**
@@ -104,7 +109,7 @@ public PutTracker createTracker(Path path, String key) {
             key,
             destKey,
             pendingsetPath,
-            owner.getWriteOperationHelper());
+            writeOperationHelper);
         LOG.debug("Created {}", tracker);
       } else {
         LOG.warn("File being created has a \"magic\" path, but the filesystem"

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

@@ -95,10 +95,4 @@ 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

@@ -20,24 +20,32 @@
 
 import javax.annotation.Nullable;
 import java.io.IOException;
+import java.nio.file.AccessDeniedException;
 import java.util.List;
 
 import com.amazonaws.AmazonClientException;
 import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsResult;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
 import com.amazonaws.services.s3.model.MultiObjectDeleteException;
 import com.amazonaws.services.s3.model.MultipartUpload;
 import com.amazonaws.services.s3.model.ObjectMetadata;
 import com.amazonaws.services.s3.model.PutObjectRequest;
 import com.amazonaws.services.s3.model.PutObjectResult;
+import com.amazonaws.services.s3.model.SelectObjectContentRequest;
+import com.amazonaws.services.s3.model.SelectObjectContentResult;
 import com.amazonaws.services.s3.model.UploadPartRequest;
 import com.amazonaws.services.s3.model.UploadPartResult;
 import com.amazonaws.services.s3.transfer.TransferManager;
-import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.fs.InvalidRequestException;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.AWSCredentialProviderList;
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.MetadataPersistenceException;
@@ -48,22 +56,17 @@
 import org.apache.hadoop.fs.s3a.auth.SignerManager;
 
 /**
- * The Guarded S3A Store.
- * This is where the core operations are implemented; the "APIs", both public
- * (FileSystem, AbstractFileSystem) and the internal ones (WriteOperationHelper)
- * call through here.
+ * The Raw S3A Store.
  */
 public interface RawS3A extends S3AService {
 
   /**
    * Bind the Raw S3 services.
-   * Ultimately the instation of some of these will be pushed down;
-   * for now they are passed in.
    * @param credentials credential source
    * @param signerManager signing
    * @param transfers bulk transfers
    * @param s3client AWS S3 client
-   * @param requestFactory
+   * @param requestFactory AWS request factory
    */
   void bind(
       StoreContext storeContext,
@@ -73,6 +76,12 @@ void bind(
       AmazonS3 s3client,
       RequestFactory requestFactory);
 
+  /**
+   * Update the S3 client.
+   * @param client new AWS S3 client.
+   */
+  void setAmazonS3Client(AmazonS3 client);
+
   /**
    * Request object metadata; increments counters in the process.
    * Retry policy: retry untranslated.
@@ -241,4 +250,56 @@ List<MultipartUpload> listMultipartUploads(String prefix)
    */
   MultipartUtils.UploadIterator listUploads(@Nullable String prefix)
       throws IOException;
+
+  /**
+ * Execute an S3 Select operation.
+ * On a failure, the request is only logged at debug to avoid the
+ * select exception being printed.
+ * @param request Select request to issue.
+ * @return response
+ * @throws IOException failure
+ */
+  @Retries.OnceRaw
+  SelectObjectContentResult select(
+    SelectObjectContentRequest request)
+    throws IOException;
+
+  /**
+   * Initiate a multipart upload from the preconfigured request.
+   * Retry policy: none + untranslated.
+   * @param request request to initiate
+   * @return the result of the call
+   * @throws AmazonClientException on failures inside the AWS SDK
+   * @throws IOException Other IO problems
+   */
+  @Retries.OnceRaw
+  InitiateMultipartUploadResult initiateMultipartUpload(
+      InitiateMultipartUploadRequest request) throws IOException;
+
+  /**
+   * Complete a multipart upload.
+   * @param request request to complete
+   * @return result
+   * @throws AmazonClientException on failures inside the AWS SDK
+   * @throws IOException Other IO problems
+   */
+  @Retries.OnceRaw
+  CompleteMultipartUploadResult completeMultipartUpload(
+      CompleteMultipartUploadRequest request) throws IOException;
+
+
+
+  /**
+   * Get the region of a bucket; fixing up the region so it can be used
+   * in the builders of other AWS clients.
+   * Requires the caller to have the AWS role permission
+   * {@code s3:GetBucketLocation}.
+   * Retry policy: retrying, translated.
+   * @param bucketName the name of the bucket
+   * @return the region in which a bucket is located
+   * @throws AccessDeniedException if the caller lacks permission.
+   * @throws IOException on any failure.
+   */
+  @Retries.RetryTranslated
+   String getBucketLocation(String bucketName) throws IOException;
 }