HADOOP-19567. Moving writing classes into impl.write package

Author: steveloughran

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

@@ -278,6 +278,7 @@ protected void serviceInit(final Configuration conf) throws Exception {
 
     // init all child services, including the stream factory
     super.serviceInit(conf);
+    // complete store writer binding
     storeWriter.bind(this, clientManager, this);
 
     // pass down extra information to the stream factory.

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

@@ -105,8 +105,7 @@ UploadInfo putObject(
    * @return the upload initiated
    * @throws SdkException on problems
    */
-  @VisibleForTesting
-  @Retries.OnceRaw("For PUT; post-PUT actions are RetryExceptionsSwallowed")
+  @Retries.OnceRaw
   PutObjectResponse putObjectDirect(PutObjectRequest putObjectRequest,
       PutObjectOptions putOptions,
       S3ADataBlocks.BlockUploadData uploadData,

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

@@ -92,6 +92,10 @@
 
 /**
  * Store Writing Operations.
+ * <p>
+ * Everything related to writing objects should go through here,
+ * including listing and aborting multipart uploads.
+ * <p>
  * The service is not ready to use until
  * {@link #bind(S3AStore, ClientManager, IORateLimiting)}
  * is invoked and the service started.
@@ -206,9 +210,10 @@ public UploadInfo putObject(
     return new UploadInfo(upload, len);
   }
 
-  @Retries.OnceRaw("For PUT; post-PUT actions are RetryExceptionsSwallowed")
   @Override
-  public PutObjectResponse putObjectDirect(PutObjectRequest putObjectRequest,
+  @Retries.OnceRaw
+  public PutObjectResponse putObjectDirect(
+      PutObjectRequest putObjectRequest,
       PutObjectOptions putOptions,
       S3ADataBlocks.BlockUploadData uploadData,
       DurationTrackerFactory trackerFactory)

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

@@ -136,7 +136,7 @@ public class WriteOperationHelper implements WriteOperations {
    * @param auditSpan span to activate
    * @param callbacks callbacks used by writeOperationHelper
    */
-  protected WriteOperationHelper(
+  public WriteOperationHelper(
       final AuditSpanSource auditSpanSource,
       final AuditSpan auditSpan,
       final WriteOperationHelperCallbacks callbacks) {
@@ -157,7 +157,10 @@ protected WriteOperationHelper(
    * @param retries number of retries
    * @param idempotent is the method idempotent
    */
-  void operationRetried(String text, Exception ex, int retries,
+  void operationRetried(
+      String text,
+      Exception ex,
+      int retries,
       boolean idempotent) {
     LOG.info("{}: Retried {}: {}", text, retries, ex.toString());
     LOG.debug("Stack", ex);
@@ -176,7 +179,8 @@ void operationRetried(String text, Exception ex, int retries,
    * @return the result of the call
    * @throws IOException any IOE raised, or translated exception
    */
-  public <T> T retry(String action,
+  public <T> T retry(
+      String action,
       String path,
       boolean idempotent,
       CallableRaisingIOE<T> operation)
@@ -285,6 +289,7 @@ private CompleteMultipartUploadResponse finalizeMultipartUpload(
       long length,
       PutObjectOptions putOptions,
       Retried retrying) throws IOException {
+
     if (partETags.isEmpty()) {
       throw new PathIOException(destKey,
           "No upload parts in multipart upload");
@@ -303,43 +308,6 @@ private CompleteMultipartUploadResponse finalizeMultipartUpload(
     }
   }
 
-  /**
-   * This completes a multipart upload to the destination key via
-   * {@code finalizeMultipartUpload()}.
-   * Retry policy: retrying, translated.
-   * Retries increment the {@code errorCount} counter.
-   * @param destKey destination
-   * @param uploadId multipart operation Id
-   * @param partETags list of partial uploads
-   * @param length length of the upload
-   * @param errorCount a counter incremented by 1 on every error; for
-   * use in statistics
-   * @param putOptions put object options
-   * @return the result of the operation.
-   * @throws IOException if problems arose which could not be retried, or
-   * the retry count was exceeded
-   */
-  @Retries.RetryTranslated
-  public CompleteMultipartUploadResponse completeMPUwithRetries(
-      String destKey,
-      String uploadId,
-      List<CompletedPart> partETags,
-      long length,
-      AtomicInteger errorCount,
-      PutObjectOptions putOptions)
-      throws IOException {
-    requireNonNull(uploadId);
-    requireNonNull(partETags);
-    LOG.debug("Completing multipart upload {} with {} parts",
-        uploadId, partETags.size());
-    return finalizeMultipartUpload(destKey,
-        uploadId,
-        partETags,
-        length,
-        putOptions,
-        (text, e, r, i) -> errorCount.incrementAndGet());
-  }
-
   /**
    * Abort a multipart upload operation.
    * @param destKey destination key of the upload
@@ -350,8 +318,11 @@ public CompleteMultipartUploadResponse completeMPUwithRetries(
    * @throws FileNotFoundException if the abort ID is unknown
    */
   @Retries.RetryTranslated
-  public void abortMultipartUpload(String destKey, String uploadId,
-      boolean shouldRetry, Retried retrying)
+  public void abortMultipartUpload(
+      String destKey,
+      String uploadId,
+      boolean shouldRetry,
+      Retried retrying)
       throws IOException {
     if (shouldRetry) {
       // retrying option
@@ -537,8 +508,8 @@ public CompleteMultipartUploadResponse commitUpload(
       throws IOException {
     requireNonNull(uploadId);
     requireNonNull(partETags);
-    LOG.debug("Completing multipart upload {} with {} parts",
-        uploadId, partETags.size());
+    LOG.debug("Committing multipart upload {} to {} with {} parts",
+        uploadId, destKey, partETags.size());
     return finalizeMultipartUpload(destKey,
         uploadId,
         partETags,
@@ -547,6 +518,44 @@ public CompleteMultipartUploadResponse commitUpload(
         Invoker.NO_OP);
   }
 
+  /**
+   * This completes a multipart upload to the destination key via
+   * {@code finalizeMultipartUpload()}.
+   * Retry policy: retrying, translated.
+   * Retries increment the {@code errorCount} counter.
+   * @param destKey destination
+   * @param uploadId multipart operation Id
+   * @param partETags list of partial uploads
+   * @param length length of the upload
+   * @param errorCount a counter incremented by 1 on every error; for
+   * use in statistics
+   * @param putOptions put object options
+   * @return the result of the operation.
+   * @throws IOException if problems arose which could not be retried, or
+   * the retry count was exceeded
+   */
+  @Retries.RetryTranslated
+  public CompleteMultipartUploadResponse completeMPUwithRetries(
+      String destKey,
+      String uploadId,
+      List<CompletedPart> partETags,
+      long length,
+      AtomicInteger errorCount,
+      PutObjectOptions putOptions)
+      throws IOException {
+    requireNonNull(uploadId);
+    requireNonNull(partETags);
+    LOG.debug("Completing multipart upload {} to {} with {} parts",
+        uploadId, destKey, partETags.size());
+    return finalizeMultipartUpload(destKey,
+        uploadId,
+        partETags,
+        length,
+        putOptions,
+        (text, e, r, i) -> errorCount.incrementAndGet());
+  }
+
+
   /**
    * Upload part of a multi-partition file.
    * @param durationTrackerFactory duration tracker factory for operation

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/write/package-info.java

@@ -18,6 +18,13 @@
 
 /**
  * Classes related to writing objects.
+ * <p>
+ * {@link org.apache.hadoop.fs.s3a.impl.write.StoreWriter} and
+ * its implementation {@link org.apache.hadoop.fs.s3a.impl.write.StoreWriterService}
+ * export the object model of S3 itself.
+ * <p>
+ * The {@code Write*} classes bridge from the filesystem APIs to the store, being
+ * invoked by filesystem, output stream and committer services.
  */
 @InterfaceAudience.Private
 package org.apache.hadoop.fs.s3a.impl.write;

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestStoreClose.java

@@ -36,7 +36,7 @@
 /**
  * Test for store closure.
  */
-public class ITestStoreClose extends AbstractS3ATestBase {Ò
+public class ITestStoreClose extends AbstractS3ATestBase {
 
   /**
    * Open a file in forced multipart, then close the fs.