HADOOP-17833. Improve Magic Committer performance (apache#3289)

Speed up the magic committer with key changes being

* Writes under __magic always retain directory markers

* File creation under __magic skips all overwrite checks,
  including the LIST call intended to stop files being
	created over dirs.
* mkdirs under __magic probes the path for existence
  but does not look any further.  	

Extra parallelism in task and job commit directory scanning
Use of createFile and openFile with parameters which all for
HEAD checks to be skipped.

The committer can write the summary _SUCCESS file to the path
`fs.s3a.committer.summary.report.directory`, which can be in a
different file system/bucket if desired, using the job id as
the filename. 

Also: HADOOP-15460. S3A FS to add `fs.s3a.create.performance`

Application code can set the createFile() option
fs.s3a.create.performance to true to disable the same
safety checks when writing under magic directories.
Use with care.

The createFile option prefix `fs.s3a.create.header.`
can be used to add custom headers to S3 objects when
created.


Contributed by Steve Loughran.

Author: steveloughran

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";
+
 }