title: ThreadpoolExecutor with retry capability and shutdown after a task fails for a number of times
body: <p>I have a need of a threadpool executor, which needs to complete an exact number (same) tasks. </p>
  
  <p>It has to be able to re-submit failed tasks for an n number of times. If any of the tasks fail for more than n, then the threadpool should shutdown and not continue to process any other tasks.</p>
  
  <p>I have tried to combine 2 approaches which I've found in different answers - one for re-submitting failed tasks by overriding ThreadPoolExecutor.afterExecute, and subclassing CountDownLatch so that threads waiting on the latch get interrupted and the executor shuts down.</p>
  
  <p>So far, this is the subclassed countdown latch:</p>
  
  <pre><code>import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.TimeUnit;
  
  public class AbortableCountDownLatch extends CountDownLatch {
  protected boolean aborted = false;
  
  public AbortableCountDownLatch(int count) {
      super(count);
  }
  
  /**
   * Unblocks all threads waiting on this latch and cause them to receive an
   * AbortedException.  If the latch has already counted all the way down,
   * this method does nothing.
   */
  public void abort() {
      if( getCount() == 0 )
          return;
  
      this.aborted = true;
      while(getCount() &gt; 0)
          countDown();
  }
  
  @Override
  public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
      final boolean rtrn = super.await(timeout,unit);
      if (aborted)
          throw new AbortedException();
      return rtrn;
  }
  
  @Override
  public void await() throws InterruptedException {
      super.await();
      if (aborted)
          throw new AbortedException();
  }
  
  
  public static class AbortedException extends InterruptedException {
      public AbortedException() {
      }
  
      public AbortedException(String detailMessage) {
          super(detailMessage);
      }
  }
  }
  </code></pre>
  
  <p>And the thread pool executor:</p>
  
  <pre><code>public class MyThreadPoolExecutor extends ThreadPoolExecutor {
  
  private static final int RETRY_LIMIT = 3;
  
  private Map&lt;Runnable, Integer&gt; retriedTasks = new ConcurrentHashMap&lt;&gt;();
  private AbortableCountDownLatch latch;
  
  public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                              TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, AbortableCountDownLatch latch) {
      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
      this.latch = latch;
  }
  
  @Override
  public void afterExecute(Runnable r, Throwable t) {
      super.afterExecute(r, t);
      // If submit() method is called instead of execute()
      if (t == null &amp;&amp; r instanceof Future&lt;?&gt;) {
          try {
              Object result = ((Future&lt;?&gt;) r).get();
          } catch (CancellationException e) {
              t = e;
          } catch (ExecutionException e) {
              t = e.getCause();
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          }
      }
      if (t != null) {
          retriedTasks.put(r, retriedTasks.getOrDefault(r, 0) + 1);
          System.out.println("Retries for " + r + " -&gt; " + retriedTasks.get(r));
          /* check to see if we have retried this task too many times, if so - shutdown */
          if (retriedTasks.containsKey(r) &amp;&amp; retriedTasks.get(r) &gt; RETRY_LIMIT) {
              System.err.println("Thread failed for more than " + RETRY_LIMIT + " times, aborting everything..");
              this.latch.abort();
          } else {
              System.err.println("Thread threw  exception " + t.getMessage() + ". Retry-ing task...");
              execute(r);
          }
      } else {
          /* clear any previous retry count for this runnable */
          retriedTasks.remove(r);
      }
  }
  }
  </code></pre>
  
  <p>And a main would be using them like this:</p>
  
  <pre><code>import java.util.Random;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  
  public class MainProcessor {
  
  public static void main(String[] args) {
  
      AbortableCountDownLatch latch = new AbortableCountDownLatch(5);
      ThreadPoolExecutor threadPoolExecutor = new MyThreadPoolExecutor(8, 8, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue&lt;&gt;(), latch);
  
      for (int i = 0; i &lt; 5; i++) {
          threadPoolExecutor.submit(() -&gt; {
              System.out.println("Started thread " + Thread.currentThread().getName());
              Random random = new Random();
              try {
                  Thread.sleep(random.nextInt(7000));
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
                  if (random.nextBoolean()){
                      System.err.println("Thread " + Thread.currentThread().getName() + " failed - throwing exception..");
                      throw new RuntimeException("Thread " + Thread.currentThread().getName() + "failed! spectacularly :!");
                  }
                  else {
                      System.out.println("Thread " + Thread.currentThread().getName() + " finished.");
                      latch.countDown();
                  }
          });
      }
  
      try {
          latch.await();
      } catch (InterruptedException e) {
          threadPoolExecutor.shutdownNow();
      }
  
      threadPoolExecutor.shutdown();
  }
  }
  </code></pre>
  
  <p>Does this approach look correct? I don't particularly like that the latch has to be passed to both the thread pool executor and to the actual Runnable. Is there a standard way of achieving this? I am fine with a Scala version too. </p>
  
  <p>I have seen others who suggest that the tasks should re-submit itself to the pool in case of failure, but that doesn't seem a good idea, as the task should only be responsible of the actual running logic, and not execution details.</p>