HADOOP-16848. Refactoring: initial layering

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/WriteOperationHelper.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,19 +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,
@@ -111,27 +105,35 @@ 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.
    */
   private final S3AStatisticsContext statisticsContext;
 
+  private final StoreContext context;
+
   /**
    * Constructor.
    * @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;
-    bucket = owner.getBucket();
+    this.context = owner.createStoreContext();
+    this.bucket = owner.getBucket();
+    this.rawS3A = rawS3A;
   }
 
   /**
@@ -227,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);
   }
 
   /**
@@ -241,14 +243,11 @@ 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());
+        () -> rawS3A.initiateMultipartUpload(initiateMPURequest).getUploadId());
   }
 
   /**
@@ -278,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(),
@@ -431,51 +428,9 @@ public UploadPartRequest newUploadPartRequest(
       InputStream uploadStream,
       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 +586,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));
   }
 
   /**
@@ -653,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/commit/MagicCommitIntegration.java

@@ -24,8 +24,8 @@
 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.WriteOperationHelper;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTracker;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 
@@ -48,21 +48,25 @@
 public class MagicCommitIntegration {
   private static final Logger LOG =
       LoggerFactory.getLogger(MagicCommitIntegration.class);
-  private final S3AFileSystem 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(S3AFileSystem 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;
   }
 
   /**
@@ -105,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/impl/AbstractS3AService.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();
+  }
+}