HADOOP-19569. put operations from store to multipart, now StoreWriter

All upload operations are in MultipartIO service, which
has been renamed and move to package org.apache.hadoop.fs.s3a.impl.write
to match.

For completeness deletion should also go into this class or an adjacent one on
deletion.

Author: steveloughran

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

@@ -1419,7 +1419,7 @@ private void initMultipartUploads(Configuration conf) throws IOException {
             Duration.ofSeconds(DEFAULT_PURGE_EXISTING_MULTIPART_AGE),
             TimeUnit.SECONDS,
             Duration.ZERO);
-        getStore().getMultipartIO().abortOutstandingMultipartUploads(
+        getStore().getStoreWriter().abortOutstandingMultipartUploads(
             purgeDuration.getSeconds(),
             "",
             maxKeys,
@@ -2626,12 +2626,12 @@ private long abortMultipartUploadsUnderPrefix(StoreContext storeContext,
     span.activate();
     // this deactivates the audit span somehow
     final RemoteIterator<MultipartUpload> uploads =
-        getStore().getMultipartIO().listMultipartUploads(storeContext, prefix, maxKeys);
+        getStore().getStoreWriter().listMultipartUploads(storeContext, prefix, maxKeys);
     // so reactivate it.
     span.activate();
     return foreach(uploads, upload ->
             invoker.retry("Aborting multipart commit", upload.key(), true, () ->
-                getStore().getMultipartIO().abortMultipartUpload(upload)));
+                getStore().getStoreWriter().abortMultipartUpload(upload)));
   }
 
   /**
@@ -3206,7 +3206,8 @@ public PutObjectRequest.Builder newPutObjectRequestBuilder(String key,
   @Retries.OnceRaw
   public UploadInfo putObject(PutObjectRequest putObjectRequest, File file,
       ProgressableProgressListener listener) throws IOException {
-    return getStore().putObject(putObjectRequest, file, listener);
+    return getStore().getStoreWriter()
+        .putObject(putObjectRequest, file, listener);
   }
 
   /**
@@ -4074,7 +4075,8 @@ PutObjectResponse executePut(
     ProgressableProgressListener listener =
         new ProgressableProgressListener(store, key, progress);
     UploadInfo info = putObject(putObjectRequest, file, listener);
-    PutObjectResponse result = getStore().waitForUploadCompletion(key, info).response();
+    PutObjectResponse result = getStore().getStoreWriter()
+        .waitForUploadCompletion(key, info).response();
     listener.uploadCompleted(info.getFileUpload());
     return result;
   }
@@ -4496,7 +4498,7 @@ private void createEmptyObject(final String objectName, PutObjectOptions putOpti
         new byte[0], 0, 0, null);
 
     invoker.retry("PUT 0-byte object ", objectName, true,
-        () -> getStore().putObjectDirect(
+        () -> getStore().getStoreWriter().putObjectDirect(
             getRequestFactory().newDirectoryMarkerRequest(objectName).build(),
             putOptions,
             uploadData,
@@ -5054,7 +5056,7 @@ public RemoteIterator<MultipartUpload> listUploads(@Nullable String prefix)
     // span is picked up retained in the listing.
     try (AuditSpan span = createSpan(MULTIPART_UPLOAD_LIST.getSymbol(),
         prefix, null)) {
-      return getStore().getMultipartIO()
+      return getStore().getStoreWriter()
           .listMultipartUploads(createStoreContext(), prefix, maxKeys);
     }
   }

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

@@ -24,7 +24,6 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.CancellationException;
 
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.exception.SdkException;
@@ -34,23 +33,18 @@
 import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
-import software.amazon.awssdk.services.s3.model.PutObjectRequest;
-import software.amazon.awssdk.services.s3.model.PutObjectResponse;
-import software.amazon.awssdk.transfer.s3.model.CompletedFileUpload;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.PathCapabilities;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
 import org.apache.hadoop.fs.s3a.impl.ClientManager;
 import org.apache.hadoop.fs.s3a.impl.MultiObjectDeleteException;
-import org.apache.hadoop.fs.s3a.impl.MultipartIO;
-import org.apache.hadoop.fs.s3a.impl.MultipartIOService;
-import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
+import org.apache.hadoop.fs.s3a.impl.StoreWriterService;
+import org.apache.hadoop.fs.s3a.impl.write.StoreWriter;
 import org.apache.hadoop.fs.s3a.impl.S3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamFactory;
@@ -118,11 +112,11 @@ public interface S3AStore extends
   ClientManager clientManager();
 
   /**
-   * Get the Multipart IO operations.
-   * @return an instance of multipart IO.
+   * Get the store writer operations.
+   * @return an instance of StoreWriterService.
    */
-  default MultipartIOService getMultipartIO() {
-    return lookupService(MultipartIO.MULTIPART_IO, MultipartIOService.class);
+  default StoreWriterService getStoreWriter() {
+    return lookupService(StoreWriter.STORE_WRITER, StoreWriterService.class);
   }
 
   /**
@@ -321,71 +315,6 @@ void deleteObjectAtPath(
       boolean isFile)
       throws SdkException, UncheckedIOException;
 
-  /**
-   * Start a transfer-manager managed async PUT of an object,
-   * incrementing the put requests and put bytes
-   * counters.
-   * <p>
-   * It does not update the other counters,
-   * as existing code does that as progress callbacks come in.
-   * Byte length is calculated from the file length, or, if there is no
-   * file, from the content length of the header.
-   * <p>
-   * Because the operation is async, any stream supplied in the request
-   * must reference data (files, buffers) which stay valid until the upload
-   * completes.
-   * Retry policy: N/A: the transfer manager is performing the upload.
-   * Auditing: must be inside an audit span.
-   * @param putObjectRequest the request
-   * @param file the file to be uploaded
-   * @param listener the progress listener for the request
-   * @return the upload initiated
-   * @throws IOException if transfer manager creation failed.
-   */
-  @Retries.OnceRaw
-  UploadInfo putObject(
-      PutObjectRequest putObjectRequest,
-      File file,
-      ProgressableProgressListener listener) throws IOException;
-
-  /**
-   * PUT an object directly (i.e. not via the transfer manager).
-   * Byte length is calculated from the file length, or, if there is no
-   * file, from the content length of the header.
-   *
-   * Retry Policy: none.
-   * Auditing: must be inside an audit span.
-   * <i>Important: this call will close any input stream in the request.</i>
-   * @param putObjectRequest the request
-   * @param putOptions put object options
-   * @param uploadData data to be uploaded
-   * @param durationTrackerFactory factory for duration tracking
-   * @return the upload initiated
-   * @throws SdkException on problems
-   */
-  @VisibleForTesting
-  @Retries.OnceRaw("For PUT; post-PUT actions are RetryExceptionsSwallowed")
-  PutObjectResponse putObjectDirect(PutObjectRequest putObjectRequest,
-      PutObjectOptions putOptions,
-      S3ADataBlocks.BlockUploadData uploadData,
-      DurationTrackerFactory durationTrackerFactory)
-      throws SdkException;
-
-  /**
-   * Wait for an upload to complete.
-   * If the upload (or its result collection) failed, this is where
-   * the failure is raised as an AWS exception.
-   * Calls {@link S3AStore#incrementPutCompletedStatistics(boolean, long)}
-   * to update the statistics.
-   * @param key destination key
-   * @param uploadInfo upload to wait for
-   * @return the upload result
-   * @throws IOException IO failure
-   * @throws CancellationException if the wait() was cancelled
-   */
-  @Retries.OnceTranslated
-  CompletedFileUpload waitForUploadCompletion(String key, UploadInfo uploadInfo)
-      throws IOException;
 
   /**
    * Get the directory allocator.

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

@@ -22,7 +22,6 @@
 
 /**
  * Interface for specific rate limiting of read and write operations.
- * {@see org.apache.hadoop.util.RateLimiting}.
  */
 public interface IORateLimiting {