HADOOP--17414. Manifest Committer -starting the tests

Based on the S3A ones; which were derived from FileOutputCommitter ones
but worked to support different destination stores.

Change-Id: I240f6d0cb55a30e66d9e84c48a7c928eead74646

Author: steveloughran

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

@@ -29,7 +29,9 @@
 
 /**
  * Abort a task.
- * This done just by deleting the task directory
+ *
+ * This done just by deleting the task directory.
+ * Exceptions may/may not be suppressed.
  */
 class AbortTaskStage extends
     AbstractJobCommitStage<Boolean, Void> {
@@ -41,8 +43,15 @@ class AbortTaskStage extends
     super(stageConfig, OP_STAGE_ABORT_TASK, false, false);
   }
 
+  /**
+   * delete the directory.
+   * @param suppressExceptions should exceptions be ignored?
+   * @return nothing
+   * @throws IOException failure when exceptions were not suppressed
+   */
   @Override
-  protected Void executeStage(final Boolean suppressExceptions) throws IOException {
+  protected Void executeStage(final Boolean suppressExceptions)
+      throws IOException {
     final Path dir = getTaskAttemptDir();
     if (dir != null) {
       LOG.info("Deleting task attempt directory {}", dir);

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

@@ -36,7 +36,6 @@
 import org.apache.hadoop.mapreduce.lib.output.committer.manifest.files.TaskManifest;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.apache.hadoop.util.DurationInfo;
-import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.util.functional.FunctionRaisingIOE;
 import org.apache.hadoop.util.functional.RemoteIterators;
 
@@ -69,44 +68,15 @@ public abstract class AbstractJobCommitStage<R, S>
       AbstractJobCommitStage.class);
 
   /**
-   * Name of the stage for statistics and logging.
-   */
-  private final String stageStatisticName;
-
-  /**
-   * IOStatistics to update.
-   */
-  private final IOStatisticsStore iostatistics;
-
-  /**
-   * Job attempt ID: never null.
-   */
-  private final String jobAttemptId;
-
-  /**
-   * ID of the task.
-   */
-  private final String taskId;
-
-  /**
-   * ID of this specific attempt at a task.
-   */
-  private final String taskAttemptId;
-
-  /**
-   * Job attempt dir.
+   * Configuration of all the stages in the ongoing committer
+   * operation.
    */
-  private final Path jobAttemptDir;
+  private final StageConfig stageConfig;
 
   /**
-   * Task attempt dir.
-   */
-  private final Path taskAttemptDir;
-
-  /**
-   * Destination of job.
+   * Name of the stage for statistics and logging.
    */
-  private final Path destinationDir;
+  private final String stageStatisticName;
 
   /**
    * Callbacks to update store.
@@ -135,11 +105,6 @@ public abstract class AbstractJobCommitStage<R, S>
    */
   private final AtomicBoolean executed = new AtomicBoolean(false);
 
-  /**
-   * Optional progress callback.
-   */
-  private final Progressable progressable;
-
   /**
    * Constructor.
    * @param stageConfig stage-independent configuration.
@@ -152,16 +117,12 @@ protected AbstractJobCommitStage(
       final String stageStatisticName,
       final boolean requireIOProcessors,
       final boolean requireManifestProcessors) {
-    this.destinationDir = requireNonNull(stageConfig.getDestinationDir());
-    this.jobAttemptDir = requireNonNull(stageConfig.getJobAttemptDir());
-    this.jobAttemptId = requireNonNull(stageConfig.getJobAttemptId());
-    this.iostatistics = requireNonNull(stageConfig.getIOStatistics());
-    this.operations = requireNonNull(stageConfig.getOperations());
-    this.progressable = stageConfig.getProgressable();
-    this.taskId = stageConfig.getTaskId();
-    this.taskAttemptId = stageConfig.getTaskAttemptId();
-    this.taskAttemptDir = stageConfig.getTaskAttemptDir();
     this.stageStatisticName = stageStatisticName;
+    this.stageConfig = stageConfig;
+    requireNonNull(stageConfig.getDestinationDir());
+    requireNonNull(stageConfig.getJobId());
+    requireNonNull(stageConfig.getJobAttemptDir());
+    this.operations = requireNonNull(stageConfig.getOperations());
     // and the processors of work if required.
     this.ioProcessors = bindProcessor(
         requireIOProcessors,
@@ -202,7 +163,7 @@ public final S apply(final R arguments) throws IOException {
     progress();
     try (DurationInfo ignored = new DurationInfo(LOG,
         true, "Executing stage %s", stageStatisticName)) {
-      return trackDuration(iostatistics, stageStatisticName, () ->
+      return trackDuration(getIOStatistics(), stageStatisticName, () ->
           executeStage(arguments));
     } finally {
       progress();
@@ -229,17 +190,22 @@ private void executeOnlyOnce() {
         "Stage attempted twice");
   }
 
+  /**
+   * The IOStatistics are shared across all uses of the
+   * StageConfig.
+   * @return the (possibly shared) IOStatistics.
+   */
   @Override
   public final IOStatisticsStore getIOStatistics() {
-    return iostatistics;
+    return stageConfig.getIOStatistics();
   }
 
   /**
    * Call progress() on any Progressable passed in.
    */
   protected final void progress() {
-    if (progressable != null) {
-      progressable.progress();
+    if (stageConfig.getProgressable() != null) {
+      stageConfig.getProgressable().progress();
     }
   }
 
@@ -266,7 +232,7 @@ protected final FileStatus getFileStatusOrNull(Path path)
    */
   protected final FileStatus getFileStatus(Path path) throws IOException {
     LOG.trace("getFileStatus('{}')", path);
-    return trackDuration(iostatistics, OP_GET_FILE_STATUS, () ->
+    return trackDuration(getIOStatistics(), OP_GET_FILE_STATUS, () ->
       operations.getFileStatus(path));
   }
 
@@ -280,7 +246,7 @@ protected final FileStatus getFileStatus(Path path) throws IOException {
   protected final boolean delete(Path path, final boolean recursive)
       throws IOException {
     LOG.trace("delete('{}, {}')", path, recursive);
-    return trackDuration(iostatistics, OP_DELETE, () ->
+    return trackDuration(getIOStatistics(), OP_DELETE, () ->
         operations.delete(path, recursive));
   }
 
@@ -293,7 +259,7 @@ protected final boolean delete(Path path, final boolean recursive)
    */
   protected final boolean mkdirs(Path path) throws IOException {
     LOG.trace("mkdirs('{}')", path);
-    return trackDuration(iostatistics, OP_MKDIRS, () ->
+    return trackDuration(getIOStatistics(), OP_MKDIRS, () ->
         operations.mkdirs(path));
   }
 
@@ -307,7 +273,7 @@ protected final boolean mkdirs(Path path) throws IOException {
   protected final RemoteIterator<FileStatus> listStatusIterator(Path path)
       throws IOException {
     LOG.trace("listStatusIterator('{}')", path);
-    return trackDuration(iostatistics, OP_LIST_STATUS, () ->
+    return trackDuration(getIOStatistics(), OP_LIST_STATUS, () ->
         operations.listStatusIterator(path));
   }
 
@@ -320,7 +286,7 @@ protected final RemoteIterator<FileStatus> listStatusIterator(Path path)
   protected final TaskManifest loadManifest(final FileStatus status)
       throws IOException {
     LOG.trace("loadManifest('{}')", status);
-    return trackDuration(iostatistics, OP_LOAD_MANIFEST, () ->
+    return trackDuration(getIOStatistics(), OP_LOAD_MANIFEST, () ->
         operations.loadTaskManifest(status));
   }
 
@@ -331,7 +297,7 @@ protected final TaskManifest loadManifest(final FileStatus status)
    */
   protected final RemoteIterator<FileStatus> listManifests() throws IOException {
     return RemoteIterators.filteringRemoteIterator(
-        listStatusIterator(jobAttemptDir),
+        listStatusIterator(getJobAttemptDir()),
         st -> st.getPath().toUri().toString().endsWith(MANIFEST_SUFFIX));
   }
 
@@ -399,7 +365,7 @@ protected final void save(AbstractManifestData manifestData,
       final Path tempPath,
       final Path finalPath) throws IOException {
     LOG.trace("save('{}, {}, {}')", manifestData, tempPath, finalPath);
-    trackDurationOfInvocation(iostatistics, OP_CREATE, () ->
+    trackDurationOfInvocation(getIOStatistics(), OP_CREATE, () ->
         operations.save(manifestData, tempPath, true));
     rename(tempPath, finalPath);
   }
@@ -416,7 +382,7 @@ protected final void rename(final Path source, final Path dest)
       throws IOException {
     // delete the destination, always.
     delete(dest, true);
-    boolean renamed = trackDuration(iostatistics, OP_RENAME_FILE, () ->
+    boolean renamed = trackDuration(getIOStatistics(), OP_RENAME_FILE, () ->
         operations.renameFile(source, dest));
     if (!renamed) {
       final FileStatus sourceStatus = getFileStatusOrNull(source);
@@ -431,17 +397,17 @@ protected final void rename(final Path source, final Path dest)
   }
 
   /**
-   * Job attempt ID: never null.
+   * Job ID: never null.
    */
-  protected String getJobAttemptId() {
-    return jobAttemptId;
+  protected String getJobId() {
+    return stageConfig.getJobId();
   }
 
   /**
    * ID of the task.
    */
   protected String getTaskId() {
-    return taskId;
+    return stageConfig.getTaskId();
   }
 
   /**
@@ -457,7 +423,7 @@ protected String getRequiredTaskId() {
    * ID of this specific attempt at a task.
    */
   protected String getTaskAttemptId() {
-    return taskAttemptId;
+    return stageConfig.getTaskAttemptId();
   }
 
   /**
@@ -474,14 +440,14 @@ protected String getRequiredTaskAttemptId() {
    * Job attempt dir.
    */
   protected Path getJobAttemptDir() {
-    return jobAttemptDir;
+    return stageConfig.getJobAttemptDir();
   }
 
   /**
    * Task attempt dir.
    */
   protected Path getTaskAttemptDir() {
-    return taskAttemptDir;
+    return stageConfig.getTaskAttemptDir();
   }
 
   /**
@@ -498,7 +464,7 @@ protected Path getRequiredTaskAttemptDir() {
    * Destination of job.
    */
   protected Path getDestinationDir() {
-    return destinationDir;
+    return stageConfig.getDestinationDir();
   }
 
   /**