code: PartitionRequestClient createPartitionRequestClient(ConnectionID connectionId) throws IOException, InterruptedException {
  		Object entry;
  		PartitionRequestClient client = null;
  
  		while (client == null) {
  			entry = clients.get(connectionId);
  
  			if (entry != null) {
  				// Existing channel or connecting channel
  				if (entry instanceof PartitionRequestClient) {
  					client = (PartitionRequestClient) entry;
  				}
  				else {
  					ConnectingChannel future = (ConnectingChannel) entry;
  					client = future.waitForChannel();
  
  					clients.replace(connectionId, future, client);
  				}
  			}
  			else {
  				// No channel yet. Create one, but watch out for a race.
  				// We create a "connecting future" and atomically add it to the map.
  				// Only the thread that really added it establishes the channel.
  				// The others need to wait on that original establisher's future.
  				ConnectingChannel connectingChannel = new ConnectingChannel(connectionId, this);
  				Object old = clients.putIfAbsent(connectionId, connectingChannel);
  
  				if (old == null) {
  					nettyClient.connect(connectionId.getAddress()).addListener(connectingChannel);
  
  					client = connectingChannel.waitForChannel();
  
  					clients.replace(connectionId, connectingChannel, client);
  				}
  				else if (old instanceof ConnectingChannel) {
  					client = ((ConnectingChannel) old).waitForChannel();
  
  					clients.replace(connectionId, old, client);
  				}
  				else {
  					client = (PartitionRequestClient) old;
  				}
  			}
  
  			// Make sure to increment the reference count before handing a client
  			// out to ensure correct bookkeeping for channel closing.
  			if (!client.incrementReferenceCounter()) {
  				destroyPartitionRequestClient(connectionId, client);
  				client = null;
  			}
  		}
  
  		return client;
  	}
docs: Atomically establishes a TCP connection to the given remote address and
  creates a {@link PartitionRequestClient} instance for this connection.