ThreadPoolExecutorWithExceptions.scala

/** Creates a fixed size thread pool executor that will rethrow exceptions from submitted jobs. */
class ThreadPoolExecutorWithExceptions(val threads: Int)
  extends ThreadPoolExecutor(threads, threads, 0, TimeUnit.SECONDS, new LinkedBlockingDeque[Runnable]) {

  /** A place to hold an exception that may have been raised by any of the executed futures. */
  protected var exception: Option[Throwable] = None

  override protected def afterExecute(runnable: Runnable, throwable: Throwable): Unit = {
    if (throwable == null && runnable.isInstanceOf[Future[_]]) try {
      val future = runnable.asInstanceOf[Future[_]]
      if (future.isDone) future.get
    } catch {
      case ex: CancellationException => exception = Some(ex)
      case ex: ExecutionException    => exception = Some(ex.getCause)
      case _:  InterruptedException  => Thread.currentThread.interrupt()
      case ex: Throwable             => exception = Some(ex)
    }
    Option(throwable).orElse(exception).foreach(throw _)
  }
}