code: @Override
    public void close() throws Exception {
      if (closed.compareAndSet(false, true)) {
        executor.shutdown();
        completionExecutor.shutdown();
        if (!executor.awaitTermination(shutdownTimeout, TimeUnit.MILLISECONDS)) {
          LOG.log(Level.WARNING, "Executor did not terminate in " + shutdownTimeout + "ms.");
          final List<Runnable> droppedRunnables = executor.shutdownNow();
          LOG.log(Level.WARNING, "Executor dropped " + droppedRunnables.size() + " tasks.");
        }
        if (!completionExecutor.awaitTermination(shutdownTimeout, TimeUnit.MILLISECONDS)) {
          LOG.log(Level.WARNING, "Executor did not terminate in " + shutdownTimeout + "ms.");
          final List<Runnable> droppedRunnables = completionExecutor.shutdownNow();
          LOG.log(Level.WARNING, "Completion executor dropped " + droppedRunnables.size() + " tasks.");
        }
      }
    }
docs: Closes the stage.