HADOOP-17833. Improve Magic Committer performance

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/audit/AuditConstants.java

@@ -90,6 +90,11 @@ private AuditConstants() {
    */
   public static final String PARAM_PROCESS = "ps";
 
+  /**
+   * Task Attempt ID query header: {@value}.
+   */
+  public static final String PARAM_TASK_ATTEMPT_ID = "ta";
+
   /**
    * Thread 0: the thread which created a span {@value}.
    */

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/audit/CommonAuditContext.java

@@ -124,11 +124,15 @@ private CommonAuditContext() {
   /**
    * Put a context entry.
    * @param key key
-   * @param value new value
+   * @param value new value., If null, triggers removal.
    * @return old value or null
    */
   public Supplier<String> put(String key, String value) {
-    return evaluatedEntries.put(key, () -> value);
+    if (value != null) {
+      return evaluatedEntries.put(key, () -> value);
+    } else {
+      return evaluatedEntries.remove(key);
+    }
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StoreStatisticNames.java

@@ -53,6 +53,9 @@ public final class StoreStatisticNames {
   /** {@value}. */
   public static final String OP_CREATE = "op_create";
 
+  /** {@value}. */
+  public static final String OP_CREATE_FILE = "op_createfile";
+
   /** {@value}. */
   public static final String OP_CREATE_NON_RECURSIVE =
       "op_create_non_recursive";

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java

@@ -297,11 +297,12 @@ public void save(FileSystem fs, Path path, T instance,
   }
 
   /**
-   * Write the JSON as bytes, then close the file.
+   * Write the JSON as bytes, then close the stream.
+   * @param instance instance to write
    * @param dataOutputStream an output stream that will always be closed
    * @throws IOException on any failure
    */
-  private void writeJsonAsBytes(T instance,
+  public void writeJsonAsBytes(T instance,
       OutputStream dataOutputStream) throws IOException {
     try {
       dataOutputStream.write(toBytes(instance));

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/functional/CloseableTaskPoolSubmitter.java

@@ -34,7 +34,7 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public final class CloseableTaskPoolSubmitter implements TaskPool.Submitter,
+public class CloseableTaskPoolSubmitter implements TaskPool.Submitter,
     Closeable {
 
   /** Executors. */

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md

@@ -26,7 +26,7 @@ create a new file or open an existing file on `FileSystem` for write.
 ## Invariants
 
 The `FSDataOutputStreamBuilder` interface does not validate parameters
-and modify the state of `FileSystem` until [`build()`](#Builder.build) is
+and modify the state of `FileSystem` until `build()` is
 invoked.
 
 ## Implementation-agnostic parameters.
@@ -110,7 +110,7 @@ of `FileSystem`.
 #### Implementation Notes
 
 The concrete `FileSystem` and/or `FSDataOutputStreamBuilder` implementation
-MUST verify that implementation-agnostic parameters (i.e., "syncable") or
+MUST verify that implementation-agnostic parameters (i.e., "syncable`) or
 implementation-specific parameters (i.e., "foofs:cache")
 are supported. `FileSystem` will satisfy optional parameters (via `opt(key, ...)`)
 on best effort. If the mandatory parameters (via `must(key, ...)`) can not be satisfied
@@ -182,3 +182,58 @@ see `FileSystem#create(path, ...)` and `FileSystem#append()`.
     result = FSDataOutputStream
 
 The result is `FSDataOutputStream` to be used to write data to filesystem.
+
+
+## <a name="s3a"></a> S3A-specific options
+
+Here are the custom options which the S3A Connector supports.
+
+| Name                        | Type      | Meaning                                |
+|-----------------------------|-----------|----------------------------------------|
+| `fs.s3a.create.performance` | `boolean` | create a file with maximum performance |
+| `fs.s3a.create.header`      | `string`  | prefix for user supplied headers       |
+
+### `fs.s3a.create.performance`
+
+Prioritize file creation performance over safety checks for filesystem consistency.
+
+This:
+1. Skips the `LIST` call which makes sure a file is being created over a directory.
+   Risk: a file is created over a directory.
+1. Ignores the overwrite flag.
+1. Never issues a `DELETE` call to delete parent directory markers.
+
+It is possible to probe an S3A Filesystem instance for this capability through
+the `hasPathCapability(path, "fs.s3a.create.performance")` check.
+
+Creating files with this option over existing directories is likely
+to make S3A filesystem clients behave inconsistently.
+
+Operations optimized for directories (e.g. listing calls) are likely
+to see the directory tree not the file; operations optimized for
+files (`getFileStatus()`, `isFile()`) more likely to see the file.
+The exact form of the inconsistencies, and which operations/parameters
+trigger this are undefined and may change between even minor releases.
+
+Using this option is the equivalent of pressing and holding down the
+"Electronic Stability Control"
+button on a rear-wheel drive car for five seconds: the safety checks are off.
+Things wil be faster if the driver knew what they were doing.
+If they didn't, the fact they had held the button down will
+be used as evidence at the inquest as proof that they made a
+conscious decision to choose speed over safety and
+that the outcome was their own fault.
+
+Accordingly: *Use if and only if you are confident that the conditions are met.*
+
+### `fs.s3a.create.header` User-supplied header support
+
+Options with the prefix `fs.s3a.create.header.` will be added to to the
+S3 object metadata as "user defined metadata".
+This metadata is visible to all applications. It can also be retrieved through the
+FileSystem/FileContext `listXAttrs()` and `getXAttrs()` API calls with the prefix `header.`
+
+When an object is renamed, the metadata is propagated the copy created.
+
+It is possible to probe an S3A Filesystem instance for this capability through
+the `hasPathCapability(path, "fs.s3a.create.header")` check.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java

@@ -29,6 +29,8 @@
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.functional.RemoteIterators;
+
 import org.junit.Assert;
 import org.junit.AssumptionViolatedException;
 import org.slf4j.Logger;
@@ -1446,11 +1448,7 @@ public static TreeScanResults treeWalk(FileSystem fs, Path path)
    */
   public static List<LocatedFileStatus> toList(
       RemoteIterator<LocatedFileStatus> iterator) throws IOException {
-    ArrayList<LocatedFileStatus> list = new ArrayList<>();
-    while (iterator.hasNext()) {
-      list.add(iterator.next());
-    }
-    return list;
+    return RemoteIterators.toList(iterator);
   }
 
   /**
@@ -1464,11 +1462,7 @@ public static List<LocatedFileStatus> toList(
    */
   public static <T extends FileStatus> List<T> iteratorToList(
           RemoteIterator<T> iterator) throws IOException {
-    List<T> list = new ArrayList<>();
-    while (iterator.hasNext()) {
-      list.add(iterator.next());
-    }
-    return list;
+    return RemoteIterators.toList(iterator);
   }
 
 

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/files/AbstractManifestData.java

@@ -70,7 +70,6 @@ public static Path unmarshallPath(String path) {
       throw new RuntimeException(
           "Failed to parse \"" + path + "\" : " + e,
           e);
-
     }
   }
 

hadoop-tools/hadoop-aws/pom.xml

@@ -405,6 +405,39 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-enforcer-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>banned-illegal-imports</id>
+            <phase>process-sources</phase>
+            <goals>
+              <goal>enforce</goal>
+            </goals>
+            <configuration>
+              <rules>
+                <restrictImports>
+                  <includeTestCode>false</includeTestCode>
+                  <reason>Restrict mapreduce imports to committer code</reason>
+                  <exclusions>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.AbstractS3ACommitter</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.AbstractS3ACommitterFactory</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.impl.*</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.magic.*</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.staging.*</exclusion>
+                  </exclusions>
+                  <bannedImports>
+                    <bannedImport>org.apache.hadoop.mapreduce.**</bannedImport>
+                    <bannedImport>org.apache.hadoop.mapred.**</bannedImport>
+                  </bannedImports>
+                </restrictImports>
+              </rules>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 

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

@@ -1159,4 +1159,22 @@ private Constants() {
    * Require that all S3 access is made through Access Points.
    */
   public static final String AWS_S3_ACCESSPOINT_REQUIRED = "fs.s3a.accesspoint.required";
+
+  /**
+   * Flag for create performance.
+   * This is *not* a configuration option; it is for use in the
+   * {code createFile()} builder.
+   * Value {@value}.
+   */
+  public static final String FS_S3A_CREATE_PERFORMANCE = "fs.s3a.create.performance";
+
+  /**
+   * Prefix for adding a header to the object when created.
+   * The actual value must have a "." suffix and then the actual header.
+   * This is *not* a configuration option; it is only for use in the
+   * {code createFile()} builder.
+   * Value {@value}.
+   */
+  public static final String FS_S3A_CREATE_HEADER = "fs.s3a.create.header";
+
 }

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

@@ -40,6 +40,7 @@
 import com.amazonaws.services.s3.model.PutObjectResult;
 import com.amazonaws.services.s3.model.UploadPartRequest;
 
+import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Futures;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ListenableFuture;
@@ -68,6 +69,7 @@
 import static org.apache.hadoop.fs.s3a.S3AUtils.*;
 import static org.apache.hadoop.fs.s3a.Statistic.*;
 import static org.apache.hadoop.fs.s3a.statistics.impl.EmptyS3AStatisticsContext.EMPTY_BLOCK_OUTPUT_STREAM_STATISTICS;
+import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfInvocation;
 import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
 
@@ -103,6 +105,11 @@ class S3ABlockOutputStream extends OutputStream implements
   /** IO Statistics. */
   private final IOStatistics iostatistics;
 
+  /**
+   * The options this instance was created with.
+   */
+  private final BlockOutputStreamBuilder builder;
+
   /** Total bytes for uploads submitted so far. */
   private long bytesSubmitted;
 
@@ -167,6 +174,7 @@ class S3ABlockOutputStream extends OutputStream implements
   S3ABlockOutputStream(BlockOutputStreamBuilder builder)
       throws IOException {
     builder.validate();
+    this.builder = builder;
     this.key = builder.key;
     this.blockFactory = builder.blockFactory;
     this.blockSize = (int) builder.blockSize;
@@ -332,6 +340,7 @@ public synchronized void write(byte[] source, int offset, int len)
    *                     initializing the upload, or if a previous operation
    *                     has failed.
    */
+  @Retries.RetryTranslated
   private synchronized void uploadCurrentBlock(boolean isLast)
       throws IOException {
     Preconditions.checkState(hasActiveBlock(), "No active block");
@@ -353,6 +362,7 @@ private synchronized void uploadCurrentBlock(boolean isLast)
    * can take time and potentially fail.
    * @throws IOException failure to initialize the upload
    */
+  @Retries.RetryTranslated
   private void initMultipartUpload() throws IOException {
     if (multiPartUpload == null) {
       LOG.debug("Initiating Multipart upload");
@@ -546,9 +556,15 @@ private int putObject() throws IOException {
     int size = block.dataSize();
     final S3ADataBlocks.BlockUploadData uploadData = block.startUpload();
     final PutObjectRequest putObjectRequest = uploadData.hasFile() ?
-        writeOperationHelper.createPutObjectRequest(key, uploadData.getFile())
-        : writeOperationHelper.createPutObjectRequest(key,
-            uploadData.getUploadStream(), size, null);
+        writeOperationHelper.createPutObjectRequest(
+            key,
+            uploadData.getFile(),
+            builder.putOptions)
+        : writeOperationHelper.createPutObjectRequest(
+            key,
+            uploadData.getUploadStream(),
+            size,
+            builder.putOptions);
     BlockUploadProgress callback =
         new BlockUploadProgress(
             block, progressListener,  now());
@@ -559,7 +575,7 @@ private int putObject() throws IOException {
           try {
             // the putObject call automatically closes the input
             // stream afterwards.
-            return writeOperationHelper.putObject(putObjectRequest);
+            return writeOperationHelper.putObject(putObjectRequest, builder.putOptions);
           } finally {
             cleanupWithLogger(LOG, uploadData, block);
           }
@@ -702,8 +718,21 @@ private class MultiPartUpload {
      */
     private IOException blockUploadFailure;
 
+    /**
+     * Constructor.
+     * Initiates the MPU request against S3.
+     * @param key upload destination
+     * @throws IOException failure
+     */
+
+    @Retries.RetryTranslated
     MultiPartUpload(String key) throws IOException {
-      this.uploadId = writeOperationHelper.initiateMultiPartUpload(key);
+      this.uploadId = trackDuration(statistics,
+          OBJECT_MULTIPART_UPLOAD_INITIATED.getSymbol(),
+          () -> writeOperationHelper.initiateMultiPartUpload(
+              key,
+              builder.putOptions));
+
       this.partETagsFutures = new ArrayList<>(2);
       LOG.debug("Initiated multi-part upload for {} with " +
           "id '{}'", writeOperationHelper, uploadId);
@@ -887,7 +916,8 @@ private void complete(List<PartETag> partETags)
                   uploadId,
                   partETags,
                   bytesSubmitted,
-                  errorCount);
+                  errorCount,
+                  builder.putOptions);
             });
       } finally {
         statistics.exceptionInMultipartComplete(errorCount.get());
@@ -1057,6 +1087,11 @@ public static final class BlockOutputStreamBuilder {
     /** is Client side Encryption enabled? */
     private boolean isCSEEnabled;
 
+    /**
+     * Put object options.
+     */
+    private PutObjectOptions putOptions;
+
     private BlockOutputStreamBuilder() {
     }
 
@@ -1070,6 +1105,7 @@ public void validate() {
       requireNonNull(statistics, "null statistics");
       requireNonNull(writeOperations, "null writeOperationHelper");
       requireNonNull(putTracker, "null putTracker");
+      requireNonNull(putOptions, "null putOptions");
       Preconditions.checkArgument(blockSize >= Constants.MULTIPART_MIN_SIZE,
           "Block size is too small: %s", blockSize);
     }
@@ -1182,5 +1218,16 @@ public BlockOutputStreamBuilder withCSEEnabled(boolean value) {
       isCSEEnabled = value;
       return this;
     }
+
+    /**
+     * Set builder value.
+     * @param value new value
+     * @return the builder
+     */
+    public BlockOutputStreamBuilder withPutOptions(
+        final PutObjectOptions value) {
+      putOptions = value;
+      return this;
+    }
   }
 }