public TransportClient createClient(String remoteHost, int remotePort)
      throws IOException, InterruptedException {
    // Get connection from the connection pool first.
    // If it is not found or not active, create a new one.
    // Use unresolved address here to avoid DNS resolution each time we creates a client.
    final InetSocketAddress unresolvedAddress =
      InetSocketAddress.createUnresolved(remoteHost, remotePort);

    // Create the ClientPool if we don't have it yet.
    ClientPool clientPool = connectionPool.get(unresolvedAddress);
    if (clientPool == null) {
      connectionPool.putIfAbsent(unresolvedAddress, new ClientPool(numConnectionsPerPeer));
      clientPool = connectionPool.get(unresolvedAddress);
    }

    int clientIndex = rand.nextInt(numConnectionsPerPeer);
    TransportClient cachedClient = clientPool.clients[clientIndex];

    if (cachedClient != null && cachedClient.isActive()) {
      // Make sure that the channel will not timeout by updating the last use time of the
      // handler. Then check that the client is still alive, in case it timed out before
      // this code was able to update things.
      TransportChannelHandler handler = cachedClient.getChannel().pipeline()
        .get(TransportChannelHandler.class);
      synchronized (handler) {
        handler.getResponseHandler().updateTimeOfLastRequest();
      }

      if (cachedClient.isActive()) {
        logger.trace("Returning cached connection to {}: {}",
          cachedClient.getSocketAddress(), cachedClient);
        return cachedClient;
      }
    }

    // If we reach here, we don't have an existing connection open. Let's create a new one.
    // Multiple threads might race here to create new connections. Keep only one of them active.
    final long preResolveHost = System.nanoTime();
    final InetSocketAddress resolvedAddress = new InetSocketAddress(remoteHost, remotePort);
    final long hostResolveTimeMs = (System.nanoTime() - preResolveHost) / 1000000;
    if (hostResolveTimeMs > 2000) {
      logger.warn("DNS resolution for {} took {} ms", resolvedAddress, hostResolveTimeMs);
    } else {
      logger.trace("DNS resolution for {} took {} ms", resolvedAddress, hostResolveTimeMs);
    }

    synchronized (clientPool.locks[clientIndex]) {
      cachedClient = clientPool.clients[clientIndex];

      if (cachedClient != null) {
        if (cachedClient.isActive()) {
          logger.trace("Returning cached connection to {}: {}", resolvedAddress, cachedClient);
          return cachedClient;
        } else {
          logger.info("Found inactive connection to {}, creating a new one.", resolvedAddress);
        }
      }
      clientPool.clients[clientIndex] = createClient(resolvedAddress);
      return clientPool.clients[clientIndex];
    }
  }
public TransportClient createClient ( String remoteHost , int remotePort ) throws IOException , InterruptedException { final InetSocketAddress unresolvedAddress = InetSocketAddress . createUnresolved ( remoteHost , remotePort ) ; ClientPool clientPool = connectionPool . get ( unresolvedAddress ) ; if ( clientPool == null ) { connectionPool . putIfAbsent ( unresolvedAddress , new ClientPool ( numConnectionsPerPeer ) ) ; clientPool = connectionPool . get ( unresolvedAddress ) ; } int clientIndex = rand . nextInt ( numConnectionsPerPeer ) ; TransportClient cachedClient = clientPool . clients [ clientIndex ] ; if ( cachedClient != null && cachedClient . isActive ( ) ) { TransportChannelHandler handler = cachedClient . getChannel ( ) . pipeline ( ) . get ( TransportChannelHandler . class ) ; synchronized ( handler ) { handler . getResponseHandler ( ) . updateTimeOfLastRequest ( ) ; } if ( cachedClient . isActive ( ) ) { logger . trace ( "Returning cached connection to {}: {}" , cachedClient . getSocketAddress ( ) , cachedClient ) ; return cachedClient ; } } final long preResolveHost = System . nanoTime ( ) ; final InetSocketAddress resolvedAddress = new InetSocketAddress ( remoteHost , remotePort ) ; final long hostResolveTimeMs = ( System . nanoTime ( ) - preResolveHost ) / 1000000 ; if ( hostResolveTimeMs > 2000 ) { logger . warn ( "DNS resolution for {} took {} ms" , resolvedAddress , hostResolveTimeMs ) ; } else { logger . trace ( "DNS resolution for {} took {} ms" , resolvedAddress , hostResolveTimeMs ) ; } synchronized ( clientPool . locks [ clientIndex ] ) { cachedClient = clientPool . clients [ clientIndex ] ; if ( cachedClient != null ) { if ( cachedClient . isActive ( ) ) { logger . trace ( "Returning cached connection to {}: {}" , resolvedAddress , cachedClient ) ; return cachedClient ; } else { logger . info ( "Found inactive connection to {}, creating a new one." , resolvedAddress ) ; } } clientPool . clients [ clientIndex ] = createClient ( resolvedAddress ) ; return clientPool . clients [ clientIndex ] ; } }
Create a {@link TransportClient} connecting to the given remote host / port.

We maintains an array of clients (size determined by spark.shuffle.io.numConnectionsPerPeer)
and randomly picks one to use. If no client was previously created in the randomly selected
spot, this function creates a new client and places it there.

Prior to the creation of a new TransportClient, we will execute all
{@link TransportClientBootstrap}s that are registered with this factory.

This blocks until a connection is successfully established and fully bootstrapped.

Concurrency: This method is safe to call from multiple threads.