cancelTasks in SubprocedurePool can hang during task error

      LOG.debug("Waiting for local region snapshots to finish.");

      int sz = futures.size();
      try {
        // Using the completion service to process the futures that finish first first.
        for (int i = 0; i < sz; i++) {
          Future<Void> f = taskPool.take();
          f.get();
          if (!futures.remove(f)) {
            LOG.warn("unexpected future" + f);
          }
          LOG.debug("Completed " + (i+1) + "/" + sz +  " local region snapshots.");
        }
        LOG.debug("Completed " + sz +  " local region snapshots.");
        return true;
      } catch (InterruptedException e) {
        LOG.warn("Got InterruptedException in SnapshotSubprocedurePool", e);
        if (!stopped) {
          Thread.currentThread().interrupt();
          throw new ForeignException("SnapshotSubprocedurePool", e);
        }
        // we are stopped so we can just exit.
      } catch (ExecutionException e) {
        if (e.getCause() instanceof ForeignException) {
          LOG.warn("Rethrowing ForeignException from SnapshotSubprocedurePool", e);
          throw (ForeignException)e.getCause();
        }
        LOG.warn("Got Exception in SnapshotSubprocedurePool", e);
        throw new ForeignException(name, e.getCause());
      } finally {
        cancelTasks();
      }

     ...
     // evict remaining tasks and futures from taskPool.
     while (!futures.isEmpty()) {
        // block to remove cancelled futures;
        LOG.warn("Removing cancelled elements from taskPool");
        futures.remove(taskPool.take());
      }

          Future<Void> f = taskPool.take();
          f.get();