HADOOP-19569. output stream close.

Took the opportunity to move all of WriteOperationHelper to calling store,
not FS, which then required all multipart IO to go there too.

Next stage of store design: multipart IO is its own service underneath
the store, so keeping the size of Store interface and Impl under control.

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IORateLimiting.java

@@ -0,0 +1,45 @@
+/*
+ * 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.io;
+
+import java.time.Duration;
+
+/**
+ * Interface for specific rate limiting of read and write operations.
+ * {@see org.apache.hadoop.util.RateLimiting}.
+ */
+public interface IORateLimiting {
+
+  /**
+   * Acquire write capacity for operations.
+   * This should be done within retry loops.
+   * @param capacity capacity to acquire.
+   * @return time spent waiting for output.
+   */
+  Duration acquireWriteCapacity(int capacity);
+
+  /**
+   * Acquire read capacity for operations.
+   * This should be done within retry loops.
+   * @param capacity capacity to acquire.
+   * @return time spent waiting for output.
+   */
+  Duration acquireReadCapacity(int capacity);
+
+}

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

@@ -25,13 +25,9 @@
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.nio.file.AccessDeniedException;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
 import java.time.Duration;
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Date;
 import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.List;
@@ -53,15 +49,10 @@
 
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
-import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.GetBucketLocationRequest;
 import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
 import software.amazon.awssdk.services.s3.model.HeadBucketResponse;
 import software.amazon.awssdk.services.s3.model.MultipartUpload;
-import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
-import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
-import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest;
 import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
 import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
@@ -146,7 +137,6 @@
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 import org.apache.hadoop.fs.s3a.impl.StoreContextBuilder;
 import org.apache.hadoop.fs.s3a.impl.StoreContextFactory;
-import org.apache.hadoop.fs.s3a.impl.UploadContentProviders;
 import org.apache.hadoop.fs.s3a.impl.CSEUtils;
 import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
@@ -253,7 +243,6 @@
 import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_OVERWRITE;
 import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_PERFORMANCE;
 import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isUnknownBucket;
-import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.CONTENT_TYPE_OCTET_STREAM;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.AP_REQUIRED_EXCEPTION;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.ARN_BUCKET_OPTION;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.DEFAULT_UPLOAD_PART_COUNT_LIMIT;
@@ -1364,9 +1353,13 @@ public FlagSet<PerformanceFlagEnum> getPerformanceFlags() {
 
   /**
    * Get the store for low-level operations.
+   * This is absolutely not for external access; it's a single method
+   * just to ease use throughout internal code.
    * @return the store the S3A FS is working through.
    */
-  private S3AStore getStore() {
+  
+  @InterfaceAudience.Private
+  public S3AStore getStore() {
     return store;
   }
 
@@ -1431,7 +1424,10 @@ private void initMultipartUploads(Configuration conf) throws IOException {
             Duration.ofSeconds(DEFAULT_PURGE_EXISTING_MULTIPART_AGE),
             TimeUnit.SECONDS,
             Duration.ZERO);
-        abortOutstandingMultipartUploads(purgeDuration.getSeconds());
+        getStore().getMultipartIO().abortOutstandingMultipartUploads(
+            purgeDuration.getSeconds(),
+            "",
+            maxKeys);
       } catch (AccessDeniedException e) {
         instrumentation.errorIgnored();
         LOG.debug("Failed to purge multipart uploads against {}," +
@@ -1440,34 +1436,6 @@ private void initMultipartUploads(Configuration conf) throws IOException {
     }
   }
 
-  /**
-   * Abort all outstanding MPUs older than a given age.
-   * @param seconds time in seconds
-   * @throws IOException on any failure, other than 403 "permission denied"
-   */
-  @Retries.RetryTranslated
-  public void abortOutstandingMultipartUploads(long seconds)
-      throws IOException {
-    Preconditions.checkArgument(seconds >= 0);
-    Instant purgeBefore =
-        Instant.now().minusSeconds(seconds);
-    LOG.debug("Purging outstanding multipart uploads older than {}",
-        purgeBefore);
-    invoker.retry("Purging multipart uploads", bucket, true,
-        () -> {
-          RemoteIterator<MultipartUpload> uploadIterator =
-              MultipartUtils.listMultipartUploads(createStoreContext(),
-                  getS3Client(), null, maxKeys);
-
-          while (uploadIterator.hasNext()) {
-            MultipartUpload upload = uploadIterator.next();
-            if (upload.initiated().compareTo(purgeBefore) < 0) {
-              abortMultipartUpload(upload);
-            }
-          }
-        });
-  }
-
   /**
    * Return the protocol scheme for the FileSystem.
    *
@@ -1934,33 +1902,6 @@ private ObjectInputStreamCallbacks createInputStreamCallbacks(
   }
 
 
-  /**
-   * Callbacks for WriteOperationHelper.
-   */
-  private final class WriteOperationHelperCallbacksImpl
-      implements WriteOperationHelper.WriteOperationHelperCallbacks {
-
-    @Override
-    @Retries.OnceRaw
-    public CompleteMultipartUploadResponse completeMultipartUpload(
-        CompleteMultipartUploadRequest request) {
-      checkRunning();
-      return getStore().completeMultipartUpload(request);
-    }
-
-    @Override
-    @Retries.OnceRaw
-    public UploadPartResponse uploadPart(
-        final UploadPartRequest request,
-        final RequestBody body,
-        final DurationTrackerFactory durationTrackerFactory)
-        throws AwsServiceException, UncheckedIOException {
-      checkRunning();
-      return getStore().uploadPart(request, body, durationTrackerFactory);
-    }
-
-  }
-
   /**
    * Create the read context for reading from the referenced file,
    * using FS state as well as the status.
@@ -2250,12 +2191,11 @@ public WriteOperationHelper getWriteOperationHelper() {
    */
   @InterfaceAudience.Private
   public WriteOperationHelper createWriteOperationHelper(AuditSpan auditSpan) {
-    return new WriteOperationHelper(this,
-        getConf(),
-        statisticsContext,
+    return new WriteOperationHelper(
         getAuditSpanSource(),
         auditSpan,
-        new WriteOperationHelperCallbacksImpl());
+        getStore().createWriteOperationHelperCallbacks()
+    );
   }
 
   /**
@@ -2710,7 +2650,7 @@ private long abortMultipartUploadsUnderPrefix(StoreContext storeContext,
     span.activate();
     return foreach(uploads, upload ->
             invoker.retry("Aborting multipart commit", upload.key(), true, () ->
-                abortMultipartUpload(upload)));
+                getStore().getMultipartIO().abortMultipartUpload(upload)));
   }
 
   /**
@@ -2942,14 +2882,7 @@ protected void incrementGauge(Statistic statistic, long count) {
    * @param ex exception.
    */
   public void operationRetried(Exception ex) {
-    if (isThrottleException(ex)) {
-      LOG.debug("Request throttled");
-      incrementStatistic(STORE_IO_THROTTLED);
-      statisticsContext.addValueToQuantiles(STORE_IO_THROTTLE_RATE, 1);
-    } else {
-      incrementStatistic(STORE_IO_RETRY);
-      incrementStatistic(IGNORED_ERRORS);
-    }
+    getStore().operationRetried(ex);
   }
 
   /**
@@ -3294,64 +3227,6 @@ public UploadInfo putObject(PutObjectRequest putObjectRequest, File file,
     return getStore().putObject(putObjectRequest, file, listener);
   }
 
-  /**
-   * 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 {
-    checkRunning();
-    long len = getPutRequestLength(putObjectRequest);
-    LOG.debug("PUT {} bytes to {}", len, putObjectRequest.key());
-    incrementPutStartStatistics(len);
-    final UploadContentProviders.BaseContentProvider provider =
-        uploadData.getContentProvider();
-    try {
-      PutObjectResponse response =
-          trackDurationOfSupplier(nonNullDurationTrackerFactory(durationTrackerFactory),
-              OBJECT_PUT_REQUESTS.getSymbol(),
-              () -> getS3Client().putObject(putObjectRequest,
-                  RequestBody.fromContentProvider(
-                      provider,
-                      provider.getSize(),
-                      CONTENT_TYPE_OCTET_STREAM)));
-      incrementPutCompletedStatistics(true, len);
-      return response;
-    } catch (SdkException e) {
-      incrementPutCompletedStatistics(false, len);
-      throw e;
-    }
-  }
-
-  /**
-   * Get the length of the PUT, verifying that the length is known.
-   * @param putObjectRequest a request bound to a file or a stream.
-   * @return the request length
-   * @throws IllegalArgumentException if the length is negative
-   */
-  private long getPutRequestLength(PutObjectRequest putObjectRequest) {
-    long len = putObjectRequest.contentLength();
-
-    Preconditions.checkState(len >= 0, "Cannot PUT object of unknown length");
-    return len;
-  }
-
   /**
    * Upload part of a multi-partition file.
    * Increments the write and put counters.
@@ -4653,23 +4528,6 @@ private CopyObjectResponse copyFile(String srcKey, String dstKey, long size,
     return response;
   }
 
-  /**
-   * Initiate a multipart upload from the preconfigured request.
-   * Retry policy: none + untranslated.
-   * @param request request to initiate
-   * @return the result of the call
-   * @throws SdkException on failures inside the AWS SDK
-   * @throws IOException Other IO problems
-   */
-  @Retries.OnceRaw
-  CreateMultipartUploadResponse initiateMultipartUpload(
-      CreateMultipartUploadRequest request) throws IOException {
-    LOG.debug("Initiate multipart upload to {}", request.key());
-    return trackDurationOfSupplier(getDurationTrackerFactory(),
-        OBJECT_MULTIPART_UPLOAD_INITIATED.getSymbol(),
-        () -> getS3Client().createMultipartUpload(request));
-  }
-
   /**
    * Perform post-write actions.
    * <p>
@@ -4719,7 +4577,7 @@ private void createEmptyObject(final String objectName, PutObjectOptions putOpti
         new byte[0], 0, 0, null);
 
     invoker.retry("PUT 0-byte object ", objectName, true,
-        () -> putObjectDirect(
+        () -> getStore().putObjectDirect(
             getRequestFactory().newDirectoryMarkerRequest(objectName).build(),
             putOptions,
             uploadData,
@@ -5299,13 +5157,7 @@ public RemoteIterator<MultipartUpload> listUploadsUnderPrefix(
       final StoreContext storeContext,
       final @Nullable String prefix)
       throws IOException {
-    // span is picked up retained in the listing.
-    String p = prefix;
-    if (prefix != null && !prefix.isEmpty() && !prefix.endsWith("/")) {
-      p = prefix + "/";
-    }
-    // duration tracking is done in iterator.
-    return MultipartUtils.listMultipartUploads(storeContext, getS3Client(), p, maxKeys);
+    return getStore().getMultipartIO().listMultipartUploads(storeContext, prefix, maxKeys);
   }
 
   /**
@@ -5321,54 +5173,9 @@ public RemoteIterator<MultipartUpload> listUploadsUnderPrefix(
   @Retries.RetryTranslated
   public List<MultipartUpload> listMultipartUploads(String prefix)
       throws IOException {
-    // add a trailing / if needed.
-    if (prefix != null && !prefix.isEmpty() && !prefix.endsWith("/")) {
-      prefix = prefix + "/";
-    }
-    String p = prefix;
-    return invoker.retry("listMultipartUploads", p, true, () -> {
-      final ListMultipartUploadsRequest request = getRequestFactory()
-          .newListMultipartUploadsRequestBuilder(p).build();
-      return trackDuration(getInstrumentation(), MULTIPART_UPLOAD_LIST.getSymbol(), () ->
-          getS3Client().listMultipartUploads(request).uploads());
-    });
-  }
-
-  /**
-   * Abort a multipart upload.
-   * Retry policy: none.
-   * @param destKey destination key
-   * @param uploadId Upload ID
-   * @throws IOException IO failure, including any uprated SdkException
-   */
-  @Retries.OnceTranslated
-  public void abortMultipartUpload(String destKey, String uploadId) throws IOException {
-    LOG.debug("Aborting multipart upload {} to {}", uploadId, destKey);
-    trackDuration(getInstrumentation(), OBJECT_MULTIPART_UPLOAD_ABORTED.getSymbol(), () ->
-        getS3Client().abortMultipartUpload(
-            getRequestFactory().newAbortMultipartUploadRequestBuilder(
-                destKey,
-                uploadId).build()));
+    return getStore().getMultipartIO().listMultipartUploads(prefix);
   }
 
-  /**
-   * Abort a multipart upload.
-   * Retry policy: none.
-   * @param upload the listed upload to abort.
-   * @throws IOException IO failure, including any uprated SdkException
-   */
-  @Retries.OnceTranslated
-  public void abortMultipartUpload(MultipartUpload upload) throws IOException {
-    String destKey = upload.key();
-    String uploadId = upload.uploadId();
-    if (LOG.isDebugEnabled()) {
-      DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-      LOG.debug("Aborting multipart upload {} to {} initiated by {} on {}",
-          uploadId, destKey, upload.initiator(),
-          df.format(Date.from(upload.initiated())));
-    }
-    abortMultipartUpload(destKey, uploadId);
-  }
 
   /**
    * Create a new instance of the committer statistics.