code: public synchronized ListenableFuture<?> waitForMinimumWorkers()
      {
          if (currentCount >= executionMinCount) {
              return immediateFuture(null);
          }
  
          SettableFuture<?> future = SettableFuture.create();
          futures.add(future);
  
          // if future does not finish in wait period, complete with an exception
          ScheduledFuture<?> timeoutTask = executor.schedule(
                  () -> {
                      synchronized (this) {
                          future.setException(new PrestoException(
                                  GENERIC_INSUFFICIENT_RESOURCES,
                                  format("Insufficient active worker nodes. Waited %s for at least %s workers, but only %s workers are active", executionMaxWait, executionMinCount, currentCount)));
                      }
                  },
                  executionMaxWait.toMillis(),
                  MILLISECONDS);
  
          // remove future if finished (e.g., canceled, timed out)
          future.addListener(() -> {
              timeoutTask.cancel(true);
              removeFuture(future);
          }, executor);
  
          return future;
      }
docs: Returns a listener that completes when the minimum number of workers for the cluster has been met.
  Note: caller should not add a listener using the direct executor, as this can delay the
  notifications for other listeners.