HADOOP-16798 committer race condition.

Rework previous commit for a more minimal design -one which reduces the diff
with the current code.

Change-Id: Ic2a1a05d29f6974eb14f9279fbb0575e3f4ddd24

Author: steveloughran

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

@@ -841,19 +841,19 @@ protected String getRole() {
   /**
    * Returns an {@link Tasks.Submitter} for parallel tasks. The number of
    * threads in the thread-pool is set by fs.s3a.committer.threads.
-   * If num-threads is 0, this will a submitter instance which will
-   * declare itself as disabled; this is used in Tasks as a cue
+   * If num-threads is 0, this will return null;
+   * this is used in Tasks as a cue
    * to switch to single-threaded execution.
    *
    * @param context the JobContext for this commit
-   * @return a submitter
+   * @return a submitter or null
    */
   protected Tasks.Submitter buildSubmitter(
       JobContext context) {
     if (getThreadCount(context) > 0) {
       return new PoolSubmitter(context);
     } else {
-      return new Tasks.DisabledSubmitter();
+      return null;
     }
   }
 
@@ -863,15 +863,14 @@ protected Tasks.Submitter buildSubmitter(
    * If num-threads is 0, this will raise an exception.
    *
    * @param context the JobContext for this commit
-   * @return an {@link ExecutorService} or null for the number of threads
+   * @param numThreads threads
+   * @return an {@link ExecutorService} for the number of threads
    */
   private synchronized ExecutorService buildThreadPool(
-      JobContext context) {
-
+      JobContext context, int numThreads) {
+    Preconditions.checkArgument(numThreads > 0,
+        "Cannot create a thread pool with no threads");
     if (threadPool == null) {
-      int numThreads = getThreadCount(context);
-      Preconditions.checkState(numThreads > 0,
-          "Cannot create a thread pool with no threads");
       LOG.debug("{}: creating thread pool of size {}", getRole(), numThreads);
       threadPool = HadoopExecutors.newFixedThreadPool(numThreads,
           new ThreadFactoryBuilder()
@@ -906,7 +905,7 @@ private int getThreadCount(final JobContext context) {
   private synchronized Future<?> submitRunnable(
       final JobContext context,
       final Runnable task) {
-    return buildThreadPool(context).submit(task);
+    return buildThreadPool(context, getThreadCount(context)).submit(task);
   }
 
   /**
@@ -917,7 +916,12 @@ private final class PoolSubmitter implements Tasks.Submitter {
 
     private final JobContext context;
 
+    private final int numThreads;
+
     private PoolSubmitter(final JobContext context) {
+      this.numThreads = getThreadCount(context);
+      Preconditions.checkArgument(numThreads > 0,
+          "Cannot create a thread pool with no threads");
       this.context = context;
     }
 
@@ -926,10 +930,6 @@ public Future<?> submit(final Runnable task) {
       return submitRunnable(context, task);
     }
 
-    @Override
-    public boolean enabled() {
-      return true;
-    }
   }
 
   /**
@@ -953,13 +953,14 @@ protected void destroyThreadPool() {
   }
 
   /**
-   * Get the submitter pool for executing the single file commit/revert
+   * Get the thread pool for executing the single file commit/revert
    * within the commit of all uploads of a single task.
-   *
-   * @return a disabled submitter
+   * This is currently null; it is here to allow the Tasks class to
+   * provide the logic for execute/revert.
+   * @return null. always.
    */
   protected final synchronized Tasks.Submitter singleThreadSubmitter() {
-    return new Tasks.DisabledSubmitter();
+    return null;
   }
 
   /**

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

@@ -75,7 +75,7 @@ public interface FailureTask<I, E extends Exception> {
    */
   public static class Builder<I> {
     private final Iterable<I> items;
-    private Submitter service = new DisabledSubmitter();
+    private Submitter service = null;
     private FailureTask<I, ?> onFailure = null;
     private boolean stopOnFailure = false;
     private boolean suppressExceptions = false;
@@ -143,7 +143,7 @@ public Builder<I> stopAbortsOnFailure() {
     }
 
     public <E extends Exception> boolean run(Task<I, E> task) throws E {
-      if (service.enabled()) {
+      if (service != null) {
         return runParallel(task);
       } else {
         return runSingleThreaded(task);
@@ -418,29 +418,6 @@ public interface Submitter {
      * @return the future of the submitted task.
      */
     Future<?> submit(Runnable task);
-
-    /**
-     * Is this submitter enabled?
-     * @return true if work can be submitted to it.
-     */
-    boolean enabled();
-  }
-
-  /**
-   * The disabled submitter declares itself as not enabled, which is
-   * used to switch Task's execution to single threaded.
-   */
-  public static class DisabledSubmitter implements Tasks.Submitter {
-
-    @Override
-    public Future<?> submit(final Runnable task) {
-      throw new UnsupportedOperationException("submitter is disabled");
-    }
-
-    @Override
-    public boolean enabled() {
-      return false;
-    }
   }
 
 }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestTasks.java

@@ -61,7 +61,7 @@ public class TestTasks extends HadoopTestBase {
 
   /**
    * Task submitter bonded to the thread pool, or
-   * disabled for the 0-thread case.IsMeAs
+   * null for the 0-thread case.
    */
   Tasks.Submitter submitter;
   private final CounterTask failingTask
@@ -126,7 +126,7 @@ public void setup() {
               .build());
       submitter = new PoolSubmitter();
     } else {
-      submitter = new Tasks.DisabledSubmitter();
+      submitter = null;
     }
 
   }
@@ -146,11 +146,8 @@ public Future<?> submit(final Runnable task) {
       return threadPool.submit(task);
     }
 
-    @Override
-    public boolean enabled() {
-      return true;
-    }
   }
+
   /**
    * create the builder.
    * @return pre-inited builder