class PartitionRequestClientFactory {

	NettyPartitionRequestClient createPartitionRequestClient(ConnectionID connectionId) throws IOException, InterruptedException {
		while (true) {
			AtomicBoolean isTheFirstOne = new AtomicBoolean(false);
			CompletableFuture<NettyPartitionRequestClient> clientFuture = clients.computeIfAbsent(connectionId, unused -> {
				isTheFirstOne.set(true);
				return new CompletableFuture<>();
			});
			if (isTheFirstOne.get()) {
				completeFromCallable(clientFuture, () -> connectWithRetries(connectionId));
			}

			final NettyPartitionRequestClient client;
			try {
				client = clientFuture.get();
			} catch (ExecutionException e) {
				throw new IOException(e);
			}
			// Make sure to increment the reference count before handing a client
			// out to ensure correct bookkeeping for channel closing.
			if (client.incrementReferenceCounter()) {
				return client;
			} else {
				destroyPartitionRequestClient(connectionId, client);
			}
		}
	}

	  PartitionRequestClientFactory(NettyClient nettyClient);
	  PartitionRequestClientFactory(NettyClient nettyClient, int retryNumber);

}