+ added feature: builder configuration of task rejection handling policy

+ added blocking retry policy
+ ConseqExecutor shutdown now blocks on work pool shutdown before shutting down admin pool

Author: q3769

README.md

@@ -262,22 +262,36 @@ Notes:
   ```jshelllanguage
   ConseqExecutor.newInstance()
   ```
+  or
+  ```jshelllanguage
+  new ConseqExecutor.Builder().build();
+  ```
 
   The concurrency can be customized:
   ```jshelllanguage
   ConseqExecutor.newInstance(10)
   ```
+  or
+  ```jshelllanguage
+  new ConseqExecutor.Builder().concurrency(10).build();
+  ```
+
 - The default work queue capacity for any executor is unlimited, which ensures the asynchronous semantics to the caller.
-  However, in the case of `ConseqExecutor` (API Style 2), the work queue capacity can be customized:
+  However, in the case of `ConseqExecutor` (API Style 2), the work queue capacity can be customized, e.g.:
   ```jshelllanguage
   ConseqExecutor conseqExecutor = new ConseqExecutor.builder().workQueueCapacity(256).build(); // default concurrency with work queue capacity of 256
   ```  
   ```jshelllanguage
   ConseqExecutor conseqExecutor = new ConseqExecutor.builder().workQueueCapacity(256).concurrency(10).build(); 
   ```
-  When the work queue is full at capacity, the caller thread will be blocked until more room becomes available. This
-  temporarily alters the asynchronous semantics and imposes "back pressure" to caller thread, which may be a desired
-  behavior in some cases.
+- When the executor work queue is full at capacity, the async task will be rejected. The default handling policy of
+  rejection is the JDK `AbortPolicy` which raises the `RejectedExecutionException`. However, in the case
+  of `ConseqExecutor` (API Style 2), the policy can be customized with any `RejectedExecutionHandler` policy, e.g.:
+  ```jshelllanguage
+  new ConseqExecutor.builder().rejectedPolicy(conseq4j.util.MorePolicies.blockingRetryPolicy()).build(); 
+  ```
+  where the caller thread will block and retry until the task is put in the work queue. This temporarily alters the
+  asynchronous semantics and imposes "back pressure" to caller thread, which may be a desired behavior in some cases.
 
 ## Full Disclosure - Asynchronous Conundrum
 

pom.xml

@@ -27,7 +27,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.github.q3769</groupId>
     <artifactId>conseq4j</artifactId>
-    <version>20230128.20230506.0</version>
+    <version>20230128.20230510.0</version>
     <packaging>jar</packaging>
     <name>conseq4j</name>
     <description>A Java concurrent API to sequence related tasks while concurring unrelated ones</description>

src/main/java/conseq4j/execute/ConseqExecutor.java

@@ -29,6 +29,7 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.ThreadSafe;
+import java.util.Map;
 import java.util.concurrent.*;
 
 /**
@@ -41,36 +42,36 @@
 public final class ConseqExecutor implements SequentialExecutor {
     private static final int DEFAULT_CONCURRENCY = Math.max(16, Runtime.getRuntime().availableProcessors());
     private static final int DEFAULT_WORK_QUEUE_CAPACITY = Integer.MAX_VALUE;
-    private final ConcurrentMap<Object, CompletableFuture<?>> sequentialExecutors;
+    private static final ThreadPoolExecutor.AbortPolicy DEFAULT_REJECTED_HANDLER = new ThreadPoolExecutor.AbortPolicy();
+    private final Map<Object, CompletableFuture<?>> sequentialExecutors = new ConcurrentHashMap<>();
+    private final ExecutorService adminThreadPool = Executors.newCachedThreadPool();
     /**
      * The worker thread pool facilitates the overall async execution, independent of the submitted tasks. Any thread
      * from the pool can be used to execute any task, regardless of sequence keys. The pool capacity decides the overall
      * max parallelism of task execution.
      */
     private final ExecutorService workerThreadPool;
-    private final ExecutorService adminThreadPool = Executors.newCachedThreadPool();
 
-    private ConseqExecutor(int concurrency, int workQueueCapacity) {
-        this.workerThreadPool =
-                workQueueCapacity == DEFAULT_WORK_QUEUE_CAPACITY ? Executors.newFixedThreadPool(concurrency) :
-                        new ThreadPoolExecutor(concurrency,
-                                concurrency,
-                                0,
-                                TimeUnit.MILLISECONDS,
-                                new ArrayBlockingQueue<>(workQueueCapacity),
-                                new BlockingRetryHandler());
-        sequentialExecutors = new ConcurrentHashMap<>(concurrency);
+    private ConseqExecutor(@Nonnull Builder builder) {
+        this(new ThreadPoolExecutor(builder.concurrency,
+                builder.concurrency,
+                0,
+                TimeUnit.MILLISECONDS,
+                builder.workQueueCapacity == DEFAULT_WORK_QUEUE_CAPACITY ? new LinkedBlockingQueue<>() :
+                        new ArrayBlockingQueue<>(builder.workQueueCapacity),
+                Executors.defaultThreadFactory(),
+                builder.rejectedPolicy));
     }
 
-    private ConseqExecutor(@Nonnull Builder builder) {
-        this(builder.concurrency, builder.workQueueCapacity);
+    private ConseqExecutor(ThreadPoolExecutor workerThreadPool) {
+        this.workerThreadPool = workerThreadPool;
     }
 
     /**
      * @return conseq executor with default concurrency
      */
     public static @Nonnull ConseqExecutor newInstance() {
-        return new ConseqExecutor(DEFAULT_CONCURRENCY, DEFAULT_WORK_QUEUE_CAPACITY);
+        return new Builder().build();
     }
 
     /**
@@ -80,7 +81,16 @@ private ConseqExecutor(@Nonnull Builder builder) {
      * @return conseq executor with given concurrency
      */
     public static @Nonnull ConseqExecutor newInstance(int concurrency) {
-        return new ConseqExecutor(concurrency, DEFAULT_WORK_QUEUE_CAPACITY);
+        return new Builder().concurrency(concurrency).build();
+    }
+
+    /**
+     * @param workerThreadPool
+     *         the worker thread pool that facilitates the overall async execution, independent of the submitted tasks.
+     * @return new ConseqExecutor instance backed by the specified workerThreadPool
+     */
+    public static ConseqExecutor from(ThreadPoolExecutor workerThreadPool) {
+        return new ConseqExecutor(workerThreadPool);
     }
 
     private static <T> T call(Callable<T> task) {
@@ -155,6 +165,15 @@ public <T> CompletableFuture<T> submit(@NonNull Callable<T> task, @NonNull Objec
     @Override
     public void shutdown() {
         this.workerThreadPool.shutdown();
+        while (true) {
+            try {
+                if (this.workerThreadPool.awaitTermination(100, TimeUnit.MILLISECONDS)) {
+                    break;
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
         this.adminThreadPool.shutdown();
     }
 
@@ -167,51 +186,21 @@ int estimateActiveExecutorCount() {
         return this.sequentialExecutors.size();
     }
 
-    static class BlockingRetryHandler implements RejectedExecutionHandler {
-        private static void blockingRetry(Runnable r, @NonNull ThreadPoolExecutor executor) {
-            if (executor.isTerminated()) {
-                return;
-            }
-            BlockingQueue<Runnable> workQueue = executor.getQueue();
-            if (workQueue.offer(r)) {
-                return;
-            }
-            boolean interrupted = false;
-            try {
-                while (true) {
-                    try {
-                        workQueue.put(r);
-                        break;
-                    } catch (InterruptedException e) {
-                        interrupted = true;
-                    }
-                }
-            } finally {
-                if (interrupted) {
-                    Thread.currentThread().interrupt();
-                }
-            }
-        }
-
-        @Override
-        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
-            blockingRetry(r, executor);
-        }
-    }
-
     /**
      * {@code ConseqExecutor} builder static inner class.
      */
     public static final class Builder {
         private int concurrency;
         private int workQueueCapacity;
+        private RejectedExecutionHandler rejectedPolicy;
 
         /**
          *
          */
         public Builder() {
             concurrency = DEFAULT_CONCURRENCY;
             workQueueCapacity = DEFAULT_WORK_QUEUE_CAPACITY;
+            rejectedPolicy = DEFAULT_REJECTED_HANDLER;
         }
 
         /**
@@ -238,6 +227,17 @@ public Builder workQueueCapacity(int workQueueCapacity) {
             return this;
         }
 
+        /**
+         * @param rejectedPolicy
+         *         handler executed by the caller thread if a task is rejected by the conseq executor, maybe e.g.
+         *         because of work queue is full. Default is {@link ThreadPoolExecutor.AbortPolicy}
+         * @return a reference to this Builder
+         */
+        public Builder rejectedPolicy(RejectedExecutionHandler rejectedPolicy) {
+            this.rejectedPolicy = rejectedPolicy;
+            return this;
+        }
+
         /**
          * Returns a {@code ConseqExecutor} built from the parameters previously set.
          *

src/main/java/conseq4j/util/MorePolicies.java

@@ -0,0 +1,62 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2021 Qingtian Wang
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package conseq4j.util;
+
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ *
+ */
+public class MorePolicies {
+    private MorePolicies() {
+    }
+
+    /**
+     * @return a {@link RejectedExecutionHandler} that uses/blocks the caller thread to retry the rejected task, or drop
+     *         the task if the executor has been terminated.
+     */
+    public static RejectedExecutionHandler blockingRetryPolicy() {
+        return new BlockingRetryPolicy();
+    }
+
+    /**
+     *
+     */
+    private static class BlockingRetryPolicy implements RejectedExecutionHandler {
+
+        @Override
+        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+            if (executor.isTerminated()) {
+                return;
+            }
+            try {
+                executor.getQueue().put(r);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+}