@Override
    public CompletableFuture<CoordinationResponse> sendCoordinationRequest(
            JobID jobId, OperatorID operatorId, CoordinationRequest request) {
        ClientCoordinationHeaders headers = ClientCoordinationHeaders.getInstance();
        ClientCoordinationMessageParameters params = new ClientCoordinationMessageParameters();
        params.jobPathParameter.resolve(jobId);
        params.operatorPathParameter.resolve(operatorId);

        SerializedValue<CoordinationRequest> serializedRequest;
        try {
            serializedRequest = new SerializedValue<>(request);
        } catch (IOException e) {
            return FutureUtils.completedExceptionally(e);
        }

        ClientCoordinationRequestBody requestBody =
                new ClientCoordinationRequestBody(serializedRequest);
        return sendRequest(headers, params, requestBody)
                .thenApply(
                        responseBody -> {
                            try {
                                return responseBody
                                        .getSerializedCoordinationResponse()
                                        .deserializeValue(getClass().getClassLoader());
                            } catch (IOException | ClassNotFoundException e) {
                                throw new CompletionException(
                                        "Failed to deserialize coordination response", e);
                            }
                        });
    }
@Test
    void testSendCoordinationRequest() throws Exception {
        final TestClientCoordinationHandler handler = new TestClientCoordinationHandler();

        try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(handler)) {
            RestClusterClient<?> restClusterClient =
                    createRestClusterClient(restServerEndpoint.getServerAddress().getPort());

            String payload = "testing payload";
            TestCoordinationRequest<String> request = new TestCoordinationRequest<>(payload);
            try {
                CompletableFuture<CoordinationResponse> future =
                        restClusterClient.sendCoordinationRequest(jobId, new OperatorID(), request);
                TestCoordinationResponse response = (TestCoordinationResponse) future.get();

                assertThat(response.payload).isEqualTo(payload);
            } finally {
                restClusterClient.close();
            }
        }
    }