HADOOP-19569. S3A: stream write/close fails badly once FS is closed

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,71 +16,77 @@
  * 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;
-
-import org.junit.AfterClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.Timeout;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+package org.apache.hadoop.util;
 
 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 static org.junit.Assert.assertEquals;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.test.AbstractHadoopTestBase;
+
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
- * Basic test for S3A's blocking executor service.
+ * Test for the blocking executor service.
  */
-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;
+
   private static final int TASK_SLEEP_MSEC = 100;
+
   private static final int SHUTDOWN_WAIT_MSEC = 200;
+
   private static final int SHUTDOWN_WAIT_TRIES = 5;
+
   private static final int BLOCKING_THRESHOLD_MSEC = 50;
 
   private static final Integer SOME_VALUE = 1337;
 
-  private static BlockingThreadPoolExecutorService tpe;
+  private BlockingThreadPoolExecutorService tpe;
+
 
-  @Rule
-  public Timeout testTimeout = new Timeout(60, TimeUnit.SECONDS);
+  @BeforeEach
+  public void setup() throws Exception {
+    ensureCreated();
+  }
 
-  @AfterClass
-  public static void afterClass() throws Exception {
+  @AfterEach
+  public void teardown() 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(v).isEqualTo(SOME_VALUE);
   }
 
   /**
    * More involved test, including detecting blocking when at capacity.
    */
   @Test
   public void testSubmitRunnable() throws Exception {
-    ensureCreated();
     verifyQueueSize(tpe, NUM_ACTIVE_TASKS + NUM_WAITING_TASKS);
   }
 
@@ -102,27 +108,30 @@ protected void verifyQueueSize(ExecutorService executorService,
     assertDidBlock(stopWatch);
   }
 
-  @Test
-  public void testShutdown() throws Exception {
-    // Cover create / destroy, regardless of when this test case runs
-    ensureCreated();
-    ensureDestroyed();
-
-    // Cover create, execute, destroy, regardless of when test case runs
-    ensureCreated();
-    testSubmitRunnable();
-    ensureDestroyed();
-  }
-
   @Test
   public void testChainedQueue() throws Throwable {
-    ensureCreated();
     int size = 2;
     ExecutorService wrapper = new SemaphoredDelegatingExecutor(tpe,
         size, true);
     verifyQueueSize(wrapper, size);
   }
 
+  @Test
+  public void testShutdownQueueRejectsOperations() throws Throwable {
+    tpe.shutdown();
+    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));
+  }
+
   // Helper functions, etc.
 
   private void assertDidBlock(StopWatch sw) {
@@ -135,28 +144,27 @@ 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("This runnable raises and exception");
   };
 
-  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;
 
     LatchedSleeper(CountDownLatch latch) {
@@ -178,7 +186,7 @@ public void run() {
   /**
    * Helper function to create thread pool under test.
    */
-  private static void ensureCreated() throws Exception {
+  private void ensureCreated() throws Exception {
     if (tpe == null) {
       LOG.debug("Creating thread pool");
       tpe = BlockingThreadPoolExecutorService.newInstance(
@@ -191,7 +199,7 @@ private static void ensureCreated() throws Exception {
    * Helper function to terminate thread pool under test, asserting that
    * shutdown -> terminate works as expected.
    */
-  private static void ensureDestroyed() throws Exception {
+  private void ensureDestroyed() throws Exception {
     if (tpe == null) {
       return;
     }