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 testConnectionReuseWhenRemoteCloseAndNoInputChannel() throws Exception {
        CompletableFuture<Void> inactiveFuture = new CompletableFuture<>();
        CompletableFuture<Channel> serverChannelFuture = new CompletableFuture<>();
        NettyProtocol protocol =
                new NettyProtocol(null, null) {
                    @Override
                    public ChannelHandler[] getServerChannelHandlers() {
                        return new ChannelHandler[] {
                            // Close on read
                            new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRegistered(ChannelHandlerContext ctx)
                                        throws Exception {
                                    super.channelRegistered(ctx);
                                    serverChannelFuture.complete(ctx.channel());
                                }
                            }
                        };
                    }

                    @Override
                    public ChannelHandler[] getClientChannelHandlers() {
                        return new ChannelHandler[] {
                            new ChannelInactiveFutureHandler(inactiveFuture)
                        };
                    }
                };
        NettyTestUtil.NettyServerAndClient serverAndClient = initServerAndClient(protocol);

        PartitionRequestClientFactory factory =
                new PartitionRequestClientFactory(
                        serverAndClient.client(), 2, 1, connectionReuseEnabled);

        ConnectionID connectionID = serverAndClient.getConnectionID(RESOURCE_ID, 0);
        NettyPartitionRequestClient oldClient = factory.createPartitionRequestClient(connectionID);

        // close server channel
        Channel channel = serverChannelFuture.get();
        channel.close();
        inactiveFuture.get();
        NettyPartitionRequestClient newClient = factory.createPartitionRequestClient(connectionID);
        assertThat(newClient).as("Factory should create a new client.").isNotSameAs(oldClient);
        shutdown(serverAndClient);
    }