Move things

* rate limiting api to hadoop-aws
* semaphored delegating excutor Itest to hadoop common unit test

Author: steveloughran

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

@@ -128,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();
@@ -140,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();
@@ -211,7 +213,11 @@ public String toString() {
     return sb.toString();
   }
 
-  private void rejectWhenShutdown() {
+  /**
+   * 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");
     }

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.After;
+import org.junit.Before;
+import org.junit.Test;
+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);
 
-  @AfterClass
-  public static void afterClass() throws Exception {
+  @Before
+  public void setup() throws Exception {
+    ensureCreated();
+  }
+
+  @After
+  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;
     }

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

@@ -57,7 +57,7 @@
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
-import org.apache.hadoop.io.IORateLimiting;
+import org.apache.hadoop.fs.s3a.api.IORateLimiting;
 import org.apache.hadoop.service.Service;
 
 /**
@@ -117,6 +117,14 @@ public interface S3AStore extends
 
   ClientManager clientManager();
 
+  /**
+   * Get the Multipart IO operations.
+   * @return an instance of multipart IO.
+   */
+  default MultipartIOService getMultipartIO() {
+    return lookupService(MultipartIO.MULTIPART_IO, MultipartIOService.class);
+  }
+
   /**
    * Look up a service by name, validate its classtype and then return the cast value.
    * This allows for the lookup of any registered service within the store, if the name
@@ -439,13 +447,6 @@ default boolean hasCapability(String capability) {
    */
   WriteOperationHelper.WriteOperationHelperCallbacks createWriteOperationHelperCallbacks();
 
-  /**
-   * Get the Multipart IO operations.
-   * @return an instance of multipart IO.
-   */
-  default MultipartIOService getMultipartIO() {
-    return lookupService(MultipartIO.MULTIPART_IO, MultipartIOService.class);
-  }
 
   /*
    =============== END WriteOperationHelperCallbacks ===============

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IORateLimiting.java renamed to hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/api/IORateLimiting.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.io;
+package org.apache.hadoop.fs.s3a.api;
 
 import java.time.Duration;
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestStoreClose.java

@@ -79,4 +79,5 @@ public void testMultipartUploadClosed() throws Throwable {
     fs.close();
     intercept(RejectedExecutionException.class, out::close);
   }
+
 }