HADOOP-19569. put operations from store to multipart, now StoreWriter

All upload operations are in MultipartIO service, which
has been renamed and move to package org.apache.hadoop.fs.s3a.impl.write
to match.

For completeness deletion should also go into this class or an adjacent one on
deletion.

Pulled out multipart IO such that
there are no back references from it to
S3AStore -the final change is to define a store
statistics class which it and other things can
use to update stats.

Executors in hadoop-common to
 - pick up shutdown of inner executor and shut themselves down.
 - semaphore executor to decrement counters in this process so that
   queue state is updated
 - semaphored delegating executor unit test in common
This stops callers being able to submit work when the inner executor has
shut down.

WriteOperationHelper
* make all calls through its callback interface, rather than given a ref to S3AFS.
* Move WriteOperationHelper callbacks to S3Store layer,

Multipart IO operations
* move nearly all Multpart IO operationss out of s3afs and into a
  new mulitpart service interface and impl
* Multipart service retrieved and invoked as appropriate
* StoreImpl stores a map of ServiceName -> service.
  with a lookupService() method in S3AStore interface, it's possible to
  retrieve services through the API just by knowing their name and type
* registering all current services this way

StoreImpl to IllegalStateException on  method invocation whene the service
isn't running.

Some methods are kept open as they do seem needed.

Author: steveloughran

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

@@ -130,21 +130,20 @@ public static BlockingThreadPoolExecutorService newInstance(
     slower than enqueueing. */
     final BlockingQueue<Runnable> workQueue =
         new LinkedBlockingQueue<>(waitingTasks + activeTasks);
+    final InnerExecutorRejection rejection = new InnerExecutorRejection();
     ThreadPoolExecutor eventProcessingExecutor =
         new ThreadPoolExecutor(activeTasks, activeTasks, keepAliveTime, unit,
             workQueue, newDaemonThreadFactory(prefixName),
-            new RejectedExecutionHandler() {
-              @Override
-              public void rejectedExecution(Runnable r,
-                  ThreadPoolExecutor executor) {
-                // This is not expected to happen.
-                LOG.error("Could not submit task to executor {}",
-                    executor.toString());
-              }
-            });
+            rejection);
     eventProcessingExecutor.allowCoreThreadTimeOut(true);
-    return new BlockingThreadPoolExecutorService(waitingTasks + activeTasks,
-        eventProcessingExecutor);
+    final BlockingThreadPoolExecutorService service =
+        new BlockingThreadPoolExecutorService(waitingTasks + activeTasks,
+            eventProcessingExecutor);
+    rejection.setDelegate((r, executor) -> {
+      service.shutdown();
+    });
+
+    return service;
   }
 
   /**
@@ -164,4 +163,28 @@ public String toString() {
         .append('}');
     return sb.toString();
   }
+
+  private static class InnerExecutorRejection implements RejectedExecutionHandler {
+
+    private RejectedExecutionHandler delegate;
+
+    private RejectedExecutionHandler getDelegate() {
+      return delegate;
+    }
+
+    private void setDelegate(final RejectedExecutionHandler delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public void rejectedExecution(Runnable r,
+        ThreadPoolExecutor executor) {
+      // This is not expected to happen.
+      LOG.error("Could not submit task to executor {}",
+          executor.toString());
+      if (getDelegate() != null) {
+        delegate.rejectedExecution(r, executor);
+      }
+    }
+  }
 }

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

@@ -31,6 +31,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -127,6 +128,7 @@ public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout,
 
   @Override
   public <T> Future<T> submit(Callable<T> task) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -139,6 +141,7 @@ public <T> Future<T> submit(Callable<T> task) {
 
   @Override
   public <T> Future<T> submit(Runnable task, T result) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -151,6 +154,7 @@ public <T> Future<T> submit(Runnable task, T result) {
 
   @Override
   public Future<?> submit(Runnable task) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -163,6 +167,7 @@ public Future<?> submit(Runnable task) {
 
   @Override
   public void execute(Runnable command) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -208,6 +213,16 @@ public String toString() {
     return sb.toString();
   }
 
+  /**
+   * Raise an exception if invoked when the executor is shut down.
+   * @throws RejectedExecutionException if the executor is shut down.
+   */
+  private void rejectWhenShutdown() throws RejectedExecutionException{
+    if (isShutdown()) {
+      throw new RejectedExecutionException("ExecutorService is shutdown");
+    }
+  }
+
   /**
    * Releases a permit after the task is executed.
    */
@@ -222,6 +237,7 @@ class RunnableWithPermitRelease implements Runnable {
     @Override
     public void run() {
       try {
+        rejectWhenShutdown();
         delegatee.run();
       } finally {
         queueingPermits.release();
@@ -244,6 +260,7 @@ class CallableWithPermitRelease<T> implements Callable<T> {
     @Override
     public T call() throws Exception {
       try {
+        rejectWhenShutdown();
         return delegatee.call();
       } finally {
         queueingPermits.release();

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestBlockingThreadPoolExecutorService.java renamed to hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestBlockingThreadPoolExecutorService.java

@@ -16,34 +16,34 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.fs.s3a;
-
-import org.apache.hadoop.util.BlockingThreadPoolExecutorService;
-import org.apache.hadoop.util.SemaphoredDelegatingExecutor;
-import org.apache.hadoop.util.StopWatch;
+package org.apache.hadoop.util;
 
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import org.apache.hadoop.test.AbstractHadoopTestBase;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
- * Basic test for S3A's blocking executor service.
+ * Test for the blocking executor service.
  */
-@Timeout(60)
-public class ITestBlockingThreadPoolExecutorService {
+public class TestBlockingThreadPoolExecutorService extends AbstractHadoopTestBase {
 
   private static final Logger LOG = LoggerFactory.getLogger(
-      ITestBlockingThreadPoolExecutorService.class);
+      TestBlockingThreadPoolExecutorService.class);
 
   private static final int NUM_ACTIVE_TASKS = 4;
   private static final int NUM_WAITING_TASKS = 2;
@@ -61,15 +61,15 @@ public static void afterClass() throws Exception {
     ensureDestroyed();
   }
 
+
   /**
    * Basic test of running one trivial task.
    */
   @Test
   public void testSubmitCallable() throws Exception {
     ensureCreated();
     Future<Integer> f = tpe.submit(callableSleeper);
-    Integer v = f.get();
-    assertEquals(SOME_VALUE, v);
+    Assertions.assertThat(f.get()).isEqualTo(SOME_VALUE);
   }
 
   /**
@@ -90,9 +90,9 @@ public void testSubmitRunnable() throws Exception {
   protected void verifyQueueSize(ExecutorService executorService,
       int expectedQueueSize) {
     CountDownLatch latch = new CountDownLatch(1);
-    for (int i = 0; i < expectedQueueSize; i++) {
-      executorService.submit(new LatchedSleeper(latch));
-    }
+    IntStream.range(0, expectedQueueSize)
+        .mapToObj(i -> new LatchedSleeper(latch))
+        .forEach(executorService::submit);
     StopWatch stopWatch = new StopWatch().start();
     latch.countDown();
     executorService.submit(sleeper);
@@ -120,6 +120,27 @@ public void testChainedQueue() throws Throwable {
     verifyQueueSize(wrapper, size);
   }
 
+  @Test
+  public void testShutdownQueueRejectsOperations() throws Throwable {
+    ensureCreated();
+    tpe.shutdown();
+    try {
+      Assertions.assertThat(tpe.isShutdown())
+          .describedAs("%s should be shutdown", tpe)
+          .isTrue();
+      // runnable
+      intercept(RejectedExecutionException.class, () ->
+          tpe.submit(failToRun));
+      // callable
+      intercept(RejectedExecutionException.class, () ->
+          tpe.submit(() -> 0));
+      intercept(RejectedExecutionException.class, () ->
+          tpe.execute(failToRun));
+    } finally {
+      tpe = null;
+    }
+  }
+
   // Helper functions, etc.
 
   private void assertDidBlock(StopWatch sw) {
@@ -132,27 +153,26 @@ private void assertDidBlock(StopWatch sw) {
     }
   }
 
-  private Runnable sleeper = new Runnable() {
-    @Override
-    public void run() {
-      String name = Thread.currentThread().getName();
-      try {
-        Thread.sleep(TASK_SLEEP_MSEC);
-      } catch (InterruptedException e) {
-        LOG.info("Thread {} interrupted.", name);
-        Thread.currentThread().interrupt();
-      }
-    }
+  private Runnable failToRun = () -> {
+    throw new RuntimeException("Failed to Run");
   };
 
-  private Callable<Integer> callableSleeper = new Callable<Integer>() {
-    @Override
-    public Integer call() throws Exception {
-      sleeper.run();
-      return SOME_VALUE;
+  private Runnable sleeper = () -> {
+    String name = Thread.currentThread().getName();
+    try {
+      Thread.sleep(TASK_SLEEP_MSEC);
+    } catch (InterruptedException e) {
+      LOG.info("Thread {} interrupted.", name);
+      Thread.currentThread().interrupt();
     }
   };
 
+  private Callable<Integer> callableSleeper = () -> {
+    sleeper.run();
+    return SOME_VALUE;
+  };
+
+
   private class LatchedSleeper implements Runnable {
     private final CountDownLatch latch;