unit: TimeUnit,
    ): T? {
      taskRunner.lock.withLock {
        val waitUntil = nanoTime + unit.toNanos(timeout)
        while (true) {
          val result = poll()
          if (result != null) return result
          if (nanoTime >= waitUntil) return null
          val editCountBefore = editCount
          yieldUntil { nanoTime >= waitUntil || editCount > editCountBefore }
        }
      }
    }

            Date expiry = new Date(System.currentTimeMillis() + 2 * MAX_WAIT_TIME);
            if (!waitUntil(expiry)) {
                throw new RuntimeException("timeout waiting for test thread to stop.");
            }
            return this;
        }

        @Override
        public boolean waitUntil(Date expiry) {
            if (isCurrentThread()) {

  @Override
  public long awaitNanos(long nanosTimeout) throws InterruptedException {
    return delegate().awaitNanos(nanosTimeout);
  }

  @Override
  public boolean awaitUntil(Date deadline) throws InterruptedException {
    return delegate().awaitUntil(deadline);
  }

  @Override
  public void signal() {
    delegate().signal();
  }

  @Override
  public void signalAll() {
    delegate().signalAll();
  }

  @Override
  public long awaitNanos(long nanosTimeout) throws InterruptedException {
    return delegate().awaitNanos(nanosTimeout);
  }

  @Override
  public boolean awaitUntil(Date deadline) throws InterruptedException {
    return delegate().awaitUntil(deadline);
  }

  @Override
  public void signal() {
    delegate().signal();
  }

  @Override
  public void signalAll() {
    delegate().signalAll();
  }

        Date timeout = shortTimeout()
        lock.lock()
        try {
            LOG.info("Waiting for test threads to complete.")
            while (!threads.isEmpty()) {
                if (!threadsChanged.awaitUntil(timeout)) {
                    failed(new IllegalStateException("Timeout waiting for test threads to complete."))
                    break;
                }
            }
            threads.each { thread ->

            if (!threads.isEmpty()) {
                logger.log "waiting for ${threads.size()} test threads to complete."
            }

            while (!threads.isEmpty()) {
                if (!condition.awaitUntil(expiry)) {
                    break;
                }
            }

            if (!threads.isEmpty()) {
                logger.log "timeout waiting for ${threads.size()} threads to complete"

        lock.lock();
        try {
            while (connection == null && running) {
                try {
                    if (!condition.awaitUntil(connectExpiry)) {
                        throw new ExecException(format("Unable to connect to the child process '%s'.\n"

        return condition.awaitNanos(nanosTimeout);
      } finally {
        lock.unlock();
      }
    }

    @Override
    public boolean awaitUntil(Date deadline) throws InterruptedException {
      lock.lock();
      try {
        return condition.awaitUntil(deadline);
      } finally {
        lock.unlock();
      }
    }

    @Override
    public void signal() {
      lock.lock();

        return condition.awaitNanos(nanosTimeout);
      } finally {
        lock.unlock();
      }
    }

    @Override
    public boolean awaitUntil(Date deadline) throws InterruptedException {
      lock.lock();
      try {
        return condition.awaitUntil(deadline);
      } finally {
        lock.unlock();
      }
    }

    @Override
    public void signal() {
      lock.lock();

        runner.client().prepareIndex().setIndex(suggester.settings().analyzer().getAnalyzerSettingsIndexName()).setSource(analyzerMapping)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL).execute().actionGet();
        suggester.settings().analyzer().init();

        SuggestSettings settings = suggester.settings();
        settings.array().add(SuggestSettings.DefaultKeys.SUPPORTED_FIELDS, field);