public <
                    M extends MessageHeaders<EmptyRequestBody, P, EmptyMessageParameters>,
                    P extends ResponseBody>
            CompletableFuture<P> sendRequest(String targetAddress, int targetPort, M messageHeaders)
                    throws IOException {
        return sendRequest(
                targetAddress,
                targetPort,
                messageHeaders,
                EmptyMessageParameters.getInstance(),
                EmptyRequestBody.getInstance());
    }
@Test
    void testRestClientClosedHandling() throws Exception {
        final Configuration config = new Configuration();
        config.set(RestOptions.IDLENESS_TIMEOUT, Duration.ofMillis(5000L));

        Socket connectionSocket = null;

        try (final ServerSocket serverSocket = new ServerSocket(0);
                final RestClient restClient =
                        new RestClient(config, EXECUTOR_EXTENSION.getExecutor())) {

            final String targetAddress = "localhost";
            final int targetPort = serverSocket.getLocalPort();

            // start server
            final CompletableFuture<Socket> socketCompletableFuture =
                    CompletableFuture.supplyAsync(
                            CheckedSupplier.unchecked(
                                    () -> NetUtils.acceptWithoutTimeout(serverSocket)));

            final CompletableFuture<EmptyResponseBody> responseFuture =
                    restClient.sendRequest(
                            targetAddress,
                            targetPort,
                            new TestMessageHeaders(),
                            EmptyMessageParameters.getInstance(),
                            EmptyRequestBody.getInstance(),
                            Collections.emptyList());

            try {
                connectionSocket = socketCompletableFuture.get(TIMEOUT, TimeUnit.SECONDS);
            } catch (TimeoutException ignored) {
                // could not establish a server connection --> see that the response failed
                socketCompletableFuture.cancel(true);
            }

            restClient.close();

            FlinkAssertions.assertThatFuture(responseFuture)
                    .eventuallyFailsWith(ExecutionException.class)
                    .withCauseInstanceOf(IOException.class);
        } finally {
            if (connectionSocket != null) {
                connectionSocket.close();
            }
        }
    }