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 testConnectionTimeout() throws Exception {
        final Configuration config = new Configuration();
        config.set(RestOptions.CONNECTION_TIMEOUT, Duration.ofMillis(1L));
        try (final RestClient restClient = new RestClient(config, Executors.directExecutor())) {
            CompletableFuture<?> future =
                    restClient.sendRequest(
                            unroutableIp,
                            80,
                            new TestMessageHeaders(),
                            EmptyMessageParameters.getInstance(),
                            EmptyRequestBody.getInstance());

            FlinkAssertions.assertThatFuture(future)
                    .eventuallyFailsWith(ExecutionException.class)
                    .withCauseInstanceOf(ConnectTimeoutException.class)
                    .extracting(Throwable::getCause, as(InstanceOfAssertFactories.THROWABLE))
                    .hasMessageContaining(unroutableIp);
        }
    }