NettyPartitionRequestClient createPartitionRequestClient(ConnectionID connectionId)
            throws IOException, InterruptedException {
        // We map the input ConnectionID to a new value to restrict the number of tcp connections
        connectionId =
                new ConnectionID(
                        connectionId.getResourceID(),
                        connectionId.getAddress(),
                        connectionId.getConnectionIndex() % maxNumberOfConnections);
        while (true) {
            final CompletableFuture<NettyPartitionRequestClient> newClientFuture =
                    new CompletableFuture<>();

            final CompletableFuture<NettyPartitionRequestClient> clientFuture =
                    clients.putIfAbsent(connectionId, newClientFuture);

            final NettyPartitionRequestClient client;

            if (clientFuture == null) {
                try {
                    client = connectWithRetries(connectionId);
                } catch (Throwable e) {
                    newClientFuture.completeExceptionally(
                            new IOException("Could not create Netty client.", e));
                    clients.remove(connectionId, newClientFuture);
                    throw e;
                }

                newClientFuture.complete(client);
            } else {
                try {
                    client = clientFuture.get();
                } catch (ExecutionException e) {
                    ExceptionUtils.rethrowIOException(ExceptionUtils.stripExecutionException(e));
                    return null;
                }
            }

            // Make sure to increment the reference count before handing a client
            // out to ensure correct bookkeeping for channel closing.
            if (client.validateClientAndIncrementReferenceCounter()) {
                return client;
            } else if (client.canBeDisposed()) {
                client.closeConnection();
            } else {
                destroyPartitionRequestClient(connectionId, client);
            }
        }
    }
@TestTemplate
    void testInterruptsNotCached() throws Exception {
        NettyTestUtil.NettyServerAndClient nettyServerAndClient = createNettyServerAndClient();
        try {
            AwaitingNettyClient nettyClient =
                    new AwaitingNettyClient(nettyServerAndClient.client());
            PartitionRequestClientFactory factory =
                    new PartitionRequestClientFactory(nettyClient, connectionReuseEnabled);

            nettyClient.awaitForInterrupts = true;
            connectAndInterrupt(factory, nettyServerAndClient.getConnectionID(RESOURCE_ID, 0));

            nettyClient.awaitForInterrupts = false;
            factory.createPartitionRequestClient(
                    nettyServerAndClient.getConnectionID(RESOURCE_ID, 0));
        } finally {
            shutdown(nettyServerAndClient);
        }
    }