lastUnit = unit;
      return null;
    }

    // No need to test these methods as they are handled by WrappingExecutorServiceTest
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) {
      throw new UnsupportedOperationException();
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {

     * @throws CancellationException {@inheritDoc}
     */
    @CanIgnoreReturnValue
    @Override
    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, TimeoutException, ExecutionException {
      return sync.get(unit.toNanos(timeout));
    }

    /*
     * Improve the documentation of when InterruptedException is thrown. Our

  val connectTimeout: Int by option(
    "--connect-timeout",
  ).help(
    "Maximum time allowed for connection (seconds)",
  ).int()
    .default(DEFAULT_TIMEOUT)

  val readTimeout: Int by option("--read-timeout")
    .help("Maximum time allowed for reading data (seconds)")
    .int()
    .default(DEFAULT_TIMEOUT)

  val callTimeout: Int by option(
    "--call-timeout",
  ).help(

abstract class ForwardingSource(
  val delegate: Source,
) : Source {
  override fun read(
    sink: Buffer,
    byteCount: Long,
  ): Long = delegate.read(sink, byteCount)

  override fun timeout() = delegate.timeout()

  override fun close() = delegate.close()

    clientListener.assertFailure(
      SocketTimeoutException::class.java,
      "timeout",
      "Read timed out",
    )
    assertThat(webSocket.close(1000, null)).isFalse()
  }

  /**
   * There's no read timeout when reading the first byte of a new frame. But as soon as we start
   * reading a frame we enable the read timeout. In this test we have the server returning the first
   * byte of a frame but no more frames.

    } catch (ExecutionException e) {
      rethrow(e);
    }
    throw new AssertionError("Unreachable");
  }

  @Override
  public V get(long timeout, TimeUnit unit)
      throws InterruptedException, ExecutionException, TimeoutException {
    try {
      super.get(timeout, unit);
    } catch (ExecutionException e) {
      rethrow(e);
    }
    throw new AssertionError("Unreachable");
  }

    private var editCount = 0

    override fun poll(): T = delegate.poll()

    override fun poll(
      timeout: Long,
      unit: TimeUnit,
    ): T? {
      return taskRunner.withLock {
        val waitUntil = nanoTime + unit.toNanos(timeout)
        while (true) {
          val result = poll()
          if (result != null) return@withLock result

        });
  }

  // TODO(cpovirk): promote to Uninterruptibles, and add untimed version
  private static void joinUninterruptibly(Thread thread, long timeout, TimeUnit unit) {
    boolean interrupted = false;
    try {
      long remainingNanos = unit.toNanos(timeout);
      long end = System.nanoTime() + remainingNanos;

      while (true) {
        try {

     * @param priority The priority.
     */
    public void setPriority(final int priority) {
        this.priority = priority;
    }

    /**
     * Sets the request timeout.
     * @param bulkRequestTimeout The request timeout.
     */
    public void setRequestTimeout(final String bulkRequestTimeout) {
        requestTimeout = bulkRequestTimeout;
    }

    /**
     * Sets the request cache size.

  public boolean isTerminated() {
    synchronized (lock) {
      return shutdown && runningTasks == 0;
    }
  }

  @Override
  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    synchronized (lock) {
      while (true) {
        if (shutdown && runningTasks == 0) {
          return true;
        } else if (nanos <= 0) {
          return false;