check(!sink.finished) { "already finished" }
      require(trailers.size != 0) { "trailers.size() == 0" }
      this.sink.trailers = trailers
    }
  }

  fun readTimeout(): Timeout = readTimeout

  fun writeTimeout(): Timeout = writeTimeout

  /** Returns a source that reads data from the peer. */
  fun getSource(): Source = source

  /**
   * Returns a sink that can be used to write data to the peer.

// validating the cluster CA against a set of pinned public keys
func RetrieveValidatedConfigInfo(cfg *kubeadmapi.Discovery, timeout time.Duration) (*clientcmdapi.Config, error) {
	return retrieveValidatedConfigInfo(nil, cfg, constants.DiscoveryRetryInterval, timeout)
}

// retrieveValidatedConfigInfo is a private implementation of RetrieveValidatedConfigInfo.

  private fun readHeader() {
    if (closed) throw IOException("closed")

    // Disable the timeout to read the first byte of a new frame.
    val b0: Int
    val timeoutBefore = source.timeout().timeoutNanos()
    source.timeout().clearTimeout()
    try {
      b0 = source.readByte() and 0xff
    } finally {
      source.timeout().timeout(timeoutBefore, TimeUnit.NANOSECONDS)
    }

    opcode = b0 and B0_MASK_OPCODE

	connRemoteLocal := remote.Connection(localHost)

	t.Run("testStreamRoundtrip", func(t *testing.T) {
		defer timeout(5 * time.Second)()
		testStreamRoundtrip(t, local, remote)
		assertNoActive(t, connRemoteLocal)
		assertNoActive(t, connLocalToRemote)
	})
	t.Run("testStreamCancel", func(t *testing.T) {
		defer timeout(5 * time.Second)()
		testStreamCancel(t, local, remote)
		assertNoActive(t, connRemoteLocal)

		driverName:                 driverName,
		prepareResourcesResponse:   prepareResourcesResponse,
		unprepareResourcesResponse: unprepareResourcesResponse,
	}
	if shouldTimeout {
		timeout := *pluginClientTimeout * 2
		fakeDRADriverGRPCServer.timeout = &timeout
	}

	drapbv1.RegisterNodeServer(s, fakeDRADriverGRPCServer)

	go func() {
		go s.Serve(l)
		<-stopCh
		s.GracefulStop()
	}()

		timeout := int(s.Config.Conntrack.TCPEstablishedTimeout.Duration / time.Second)
		if err := ct.SetTCPEstablishedTimeout(ctx, timeout); err != nil {
			return err
		}
	}

	if s.Config.Conntrack.TCPCloseWaitTimeout != nil && s.Config.Conntrack.TCPCloseWaitTimeout.Duration > 0 {
		timeout := int(s.Config.Conntrack.TCPCloseWaitTimeout.Duration / time.Second)

	PeerURL string
}

func (c *Client) listMembers(timeout time.Duration) (*clientv3.MemberListResponse, error) {
	// Gets the member list
	var lastError error
	var resp *clientv3.MemberListResponse
	if timeout == 0 {
		timeout = constants.EtcdAPICallTimeout
	}
	err := wait.PollUntilContextTimeout(context.Background(), constants.EtcdAPICallRetryInterval, timeout,
		true, func(_ context.Context) (bool, error) {

    }

    private BlockingHandler addBlockingOptionalHandler(final int concurrent, final Collection<? extends ResourceExpectation> expectations) {
        return handler.addHandler(previous -> new MaybeNConcurrentRequests(lock, serverId, timeout, concurrent, previous, expectations));
    }

    /**

	Notify(cs, syscall.SIGINT)
	defer Stop(cs)

	const execs = 10
	for i := 0; i < execs; i++ {
		timeout := "0"
		if deadline, ok := t.Deadline(); ok {
			timeout = time.Until(deadline).String()
		}
		cmd := testenv.Command(t, os.Args[0], "-test.run=^TestAtomicStop$", "-test.timeout="+timeout)
		cmd.Env = append(os.Environ(), "GO_TEST_ATOMIC_STOP=1")
		out, err := cmd.CombinedOutput()
		if err == nil {

		}
		switch short {
		case "timeout":
			timeoutSet = true
		case "outputdir":
			outputDirSet = true
		}
	})

	// 'go test' has a default timeout, but the test binary itself does not.
	// If the timeout wasn't set (and forwarded) explicitly, add the default
	// timeout to the command line.
	if testTimeout > 0 && !timeoutSet {