HADOOP-19569. stream close

Continuing pulling out multipart IO such that
there are no back references from it to
S3AStore -the final change is to define a store
statistics class which it and other things can
use to update stats.
Yes, those stats will get back into the FS, but
we don't want to have that recursive complexity
of S3AFS utility classes.

Once an inner class has been pulled out of S3AStoreImpl,
it SHOULD have a restricted interface of operations it
can call back on the store.

Author: steveloughran

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

@@ -1422,7 +1422,8 @@ private void initMultipartUploads(Configuration conf) throws IOException {
         getStore().getMultipartIO().abortOutstandingMultipartUploads(
             purgeDuration.getSeconds(),
             "",
-            maxKeys);
+            maxKeys,
+            createStoreContext());
       } catch (AccessDeniedException e) {
         instrumentation.errorIgnored();
         LOG.debug("Failed to purge multipart uploads against {}," +
@@ -3208,38 +3209,6 @@ public UploadInfo putObject(PutObjectRequest putObjectRequest, File file,
     return getStore().putObject(putObjectRequest, file, listener);
   }
 
-  /**
-   * At the start of a put/multipart upload operation, update the
-   * relevant counters.
-   *
-   * @param bytes bytes in the request.
-   */
-  protected void incrementPutStartStatistics(long bytes) {
-    getStore().incrementPutStartStatistics(bytes);
-  }
-
-  /**
-   * At the end of a put/multipart upload operation, update the
-   * relevant counters and gauges.
-   *
-   * @param success did the operation succeed?
-   * @param bytes bytes in the request.
-   */
-  protected void incrementPutCompletedStatistics(boolean success, long bytes) {
-    getStore().incrementPutCompletedStatistics(success, bytes);
-  }
-
-  /**
-   * Callback for use in progress callbacks from put/multipart upload events.
-   * Increments those statistics which are expected to be updated during
-   * the ongoing upload operation.
-   * @param key key to file that is being written (for logging)
-   * @param bytes bytes successfully uploaded.
-   */
-  protected void incrementPutProgressStatistics(String key, long bytes) {
-    getStore().incrementPutProgressStatistics(key, bytes);
-  }
-
   /**
    * Delete a list of keys on a s3-backend.
    * Retry policy: retry untranslated; delete considered idempotent.
@@ -4532,7 +4501,7 @@ private void createEmptyObject(final String objectName, PutObjectOptions putOpti
             putOptions,
             uploadData,
             getDurationTrackerFactory()));
-    incrementPutProgressStatistics(objectName, 0);
+    incrementWriteOperations();
     instrumentation.directoryCreated();
   }
 

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

@@ -208,7 +208,7 @@ default MultipartIOService getMultipartIO() {
 
   /**
    * Given a possibly null duration tracker factory, return a non-null
-   * one for use in tracking durations -either that or the FS tracker
+   * one for use in tracking durations -either that or the store tracker
    * itself.
    *
    * @param factory factory.

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

@@ -109,13 +109,16 @@ AbortMultipartUploadResponse abortMultipartUpload(String destKey, String uploadI
       throws AwsServiceException;
 
   /**
-   * Listing and abort all multipart uploads.
+   * List and abort all multipart uploads older than a specified age.
+   * @param seconds age of multiparts to abort.
    * @param prefix prefix to scan for, "" for none
-   * @return a listing of multipart uploads.
+   * @param maxKeys maximum number of keys to list and abort
+   * @param context store context to use
    * @throws IOException IO failure, including any uprated SdkException
    */
   @Retries.RetryTranslated
-  void abortOutstandingMultipartUploads(long seconds, @Nullable String prefix, int maxKeys)
+  void abortOutstandingMultipartUploads(long seconds, @Nullable String prefix, int maxKeys,
+      StoreContext context)
       throws IOException;
 
   /**

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

@@ -49,12 +49,14 @@
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.S3AStore;
+import org.apache.hadoop.fs.s3a.api.IORateLimiting;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.store.audit.AuditSpan;
 import org.apache.hadoop.service.AbstractService;
 import org.apache.hadoop.util.Preconditions;
 
+import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.s3a.Statistic.MULTIPART_UPLOAD_LIST;
 import static org.apache.hadoop.fs.s3a.Statistic.MULTIPART_UPLOAD_PART_PUT;
 import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_MULTIPART_UPLOAD_ABORTED;
@@ -63,25 +65,69 @@
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfOperation;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfSupplier;
+import static org.apache.hadoop.util.Preconditions.checkArgument;
 
+/**
+ * All Multipart IO Operations.
+ * The service is not ready to use until {@link #bind(S3AStore, ClientManager, IORateLimiting)}
+ * is invoked and the service started.
+ */
 public class MultipartIOService extends AbstractService
     implements MultipartIO {
 
   private static final Logger LOG =
       LoggerFactory.getLogger(MultipartIOService.class);
 
+  /**
+   * Store for some statistics invocations.
+   */
   private S3AStore store;
 
+  /**
+   * Rate limiter (likely to be the store)
+   */
+  private IORateLimiting limiting;
+
+  /**
+   * SDK client.
+   */
   private ClientManager clientManager;
 
+  /**
+   * Create the Service with the service name {@link #MULTIPART_IO}.
+   */
   public MultipartIOService() {
-    super(MULTIPART_IO);
+    this(MULTIPART_IO);
   }
 
+  /**
+   * Constructor.
+   * @param name service name
+   */
+  public MultipartIOService(final String name) {
+    super(name);
+  }
 
-  public void bind(final S3AStore store, ClientManager clientManager) {
-    this.store = store;
+  /**
+   * Bind to dependencies.
+   * This MUST be called before service start
+   * @param store store
+   * @param clientManager sdk client manager
+   * @param limiting rate limiting.
+   */
+  public void bind(final S3AStore store,
+      ClientManager clientManager,
+      IORateLimiting limiting) {
+    this.store = requireNonNull(store);
     this.clientManager = clientManager;
+    this.limiting = store;
+  }
+
+
+  @Override
+  protected void serviceStart() throws Exception {
+    super.serviceStart();
+    requireNonNull(store);
   }
 
   /**
@@ -104,7 +150,6 @@ public CreateMultipartUploadResponse initiateMultipartUpload(
         () -> getS3ClientUnchecked().createMultipartUpload(request));
   }
 
-
   @Retries.OnceRaw
   @Override
   public UploadPartResponse uploadPart(
@@ -158,26 +203,24 @@ public AbortMultipartUploadResponse abortMultipartUpload(String destKey, String
     checkRunning();
     LOG.debug("Aborting multipart upload {} to {}", uploadId, destKey);
     return trackDurationOfSupplier(store.getInstrumentation(), OBJECT_MULTIPART_UPLOAD_ABORTED.getSymbol(), () -> {
-      store.acquireWriteCapacity(InternalConstants.MULTIPART_ABORT_WRITE_CAPACITY);
+      limiting.acquireWriteCapacity(InternalConstants.MULTIPART_ABORT_WRITE_CAPACITY);
       return getS3ClientUnchecked().abortMultipartUpload(
           store.getRequestFactory().newAbortMultipartUploadRequestBuilder(
               destKey,
               uploadId).build());
     });
   }
 
-
   @Retries.RetryTranslated
   @Override
-  public void abortOutstandingMultipartUploads(long seconds, @Nullable String prefix, int maxKeys)
+  public void abortOutstandingMultipartUploads(long seconds, @Nullable String prefix, int maxKeys,
+      StoreContext context)
       throws IOException {
-    Preconditions.checkArgument(seconds >= 0);
+    checkArgument(seconds >= 0);
     checkRunning();
-    Instant purgeBefore =
-        Instant.now().minusSeconds(seconds);
+    Instant purgeBefore = Instant.now().minusSeconds(seconds);
     LOG.debug("Purging outstanding multipart uploads older than {}",
         purgeBefore);
-    final StoreContext context = store.createStoreContext();
     context.getInvoker().retry("Purging multipart uploads",
         context.getBucket(), true,
         () -> {
@@ -208,7 +251,7 @@ public List<MultipartUpload> listMultipartUploads(@Nullable String prefix)
       final ListMultipartUploadsRequest request = store.getRequestFactory()
           .newListMultipartUploadsRequestBuilder(p).build();
       return trackDuration(store.getInstrumentation(), MULTIPART_UPLOAD_LIST.getSymbol(), () -> {
-        store.acquireReadCapacity(InternalConstants.MULTIPART_LIST_READ_CAPACITY);
+        limiting.acquireReadCapacity(InternalConstants.MULTIPART_LIST_READ_CAPACITY);
         return getS3ClientUnchecked().listMultipartUploads(request).uploads();
       });
     });
@@ -258,7 +301,6 @@ public class ListingIterator implements
 
     private final int maxKeys;
 
-
     private final Invoker invoker;
 
     private final AuditSpan auditSpan;
@@ -345,8 +387,8 @@ private void requestNextBatch() throws IOException {
       checkRunning();
 
       try (AuditSpan span = auditSpan.activate()) {
-        ListMultipartUploadsRequest.Builder requestBuilder = requestFactory
-            .newListMultipartUploadsRequestBuilder(prefix);
+        ListMultipartUploadsRequest.Builder requestBuilder =
+            requestFactory.newListMultipartUploadsRequestBuilder(prefix);
         if (!firstListing) {
           requestBuilder.keyMarker(listing.nextKeyMarker());
           requestBuilder.uploadIdMarker(listing.nextUploadIdMarker());
@@ -364,7 +406,7 @@ private void requestNextBatch() throws IOException {
             trackDurationOfOperation(storeContext.getInstrumentation(),
                 OBJECT_MULTIPART_UPLOAD_LIST.getSymbol(),
                 () -> {
-                  store.acquireReadCapacity(InternalConstants.MULTIPART_LIST_READ_CAPACITY);
+                  limiting.acquireReadCapacity(InternalConstants.MULTIPART_LIST_READ_CAPACITY);
                   return getS3ClientUnchecked().listMultipartUploads(requestBuilder.build());
                 }));
         LOG.debug("Listing found {} upload(s)",
@@ -375,7 +417,7 @@ private void requestNextBatch() throws IOException {
   }
 
   private S3Client getS3ClientUnchecked() {
-    return store.getOrCreateS3ClientUnchecked();
+    return clientManager.getOrCreateS3ClientUnchecked();
   }
 
   /**