@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 testSendCoordinationRequestException() throws Exception {
        final TestClientCoordinationHandler handler =
                new TestClientCoordinationHandler(new FlinkJobNotFoundException(jobId));
        try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(handler)) {
            try (RestClusterClient<?> restClusterClient =
                    createRestClusterClient(restServerEndpoint.getServerAddress().getPort())) {
                String payload = "testing payload";
                TestCoordinationRequest<String> request = new TestCoordinationRequest<>(payload);

                assertThatThrownBy(
                                () ->
                                        restClusterClient
                                                .sendCoordinationRequest(
                                                        jobId, new OperatorID(), request)
                                                .get())
                        .matches(
                                e ->
                                        ExceptionUtils.findThrowableWithMessage(
                                                        e,
                                                        FlinkJobNotFoundException.class.getName())
                                                .isPresent());
            }
        }
    }