class TimedCallable implements Callable<V> {

  public static <V> List<V> run(final String activity, final Logger logger, final List<TimedCallable<V>> tasks, int parallelism) throws IOException {
    Preconditions.checkArgument(!Preconditions.checkNotNull(tasks).isEmpty(), "list of tasks is empty");
    Preconditions.checkArgument(parallelism > 0);
    parallelism = Math.min(parallelism, tasks.size());
    final ExecutorService threadPool = parallelism == 1 ? MoreExecutors.newDirectExecutorService()
        : Executors.newFixedThreadPool(parallelism, new ThreadFactoryBuilder().setNameFormat(activity + "-%d").build());
    final long timeout = TIMEOUT_PER_RUNNABLE_IN_MSECS * ((tasks.size() - 1)/parallelism + 1);
    final FutureMapper<V> futureMapper = new FutureMapper<>();
    final Statistics<V> statistics = logger.isDebugEnabled() ? new Statistics<>() : null;
    try {
      return Collectors.toList(threadPool.invokeAll(tasks, timeout, TimeUnit.MILLISECONDS), futureMapper);
    } catch (InterruptedException e) {
      final String errMsg = String.format("Interrupted while waiting for activity '%s' tasks to be done.", activity);
      logger.error(errMsg, e);
      throw UserException.resourceError(e)
          .message(errMsg)
          .build(logger);
    } catch (RejectedExecutionException e) {
      final String errMsg = String.format("Failure while submitting activity '%s' tasks for execution.", activity);
      logger.error(errMsg, e);
      throw UserException.internalError(e)
          .message(errMsg)
          .build(logger);
    } finally {
      List<Runnable> notStartedTasks = threadPool.shutdownNow();
      if (!notStartedTasks.isEmpty()) {
        logger.error("{} activity '{}' tasks never commenced execution.", notStartedTasks.size(), activity);
      }
      try {
        // Wait for 5s for currently running threads to terminate. Above call (threadPool.shutdownNow()) interrupts
        // any running threads. If the tasks are handling the interrupts properly they should be able to
        // wrap up and terminate. If not waiting for 5s here gives a chance to identify and log any potential
        // thread leaks.
        if (!threadPool.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
          logger.error("Detected run away tasks in activity '{}'.", activity);
        }
      } catch (final InterruptedException e) {
        logger.warn("Interrupted while waiting for pending threads in activity '{}' to terminate.", activity);
      }

      if (statistics != null) {
        statistics.collect(tasks).log(activity, logger, parallelism);
      }
      if (futureMapper.count != tasks.size()) {
        final String errMsg = String.format("Waited for %d ms, but only %d tasks for '%s' are complete." +
            " Total number of tasks %d, parallelism %d.", timeout, futureMapper.count, activity, tasks.size(), parallelism);
        logger.error(errMsg, futureMapper.throwable);
        throw UserException.resourceError(futureMapper.throwable)
            .message(errMsg)
            .build(logger);
      }
      if (futureMapper.throwable != null) {
        throw (futureMapper.throwable instanceof IOException) ?
            (IOException)futureMapper.throwable : new IOException(futureMapper.throwable);
      }
    }
  }

}