@Override
    public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) {
        final SavepointDisposalRequest savepointDisposalRequest =
                new SavepointDisposalRequest(savepointPath);

        final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture =
                sendRequest(
                        SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest);

        final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture =
                savepointDisposalTriggerFuture.thenCompose(
                        (TriggerResponse triggerResponse) -> {
                            final TriggerId triggerId = triggerResponse.getTriggerId();
                            final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders =
                                    SavepointDisposalStatusHeaders.getInstance();
                            final SavepointDisposalStatusMessageParameters
                                    savepointDisposalStatusMessageParameters =
                                            savepointDisposalStatusHeaders
                                                    .getUnresolvedMessageParameters();
                            savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(
                                    triggerId);

                            return pollResourceAsync(
                                    () ->
                                            sendRequest(
                                                    savepointDisposalStatusHeaders,
                                                    savepointDisposalStatusMessageParameters));
                        });

        return savepointDisposalFuture.thenApply(
                (AsynchronousOperationInfo asynchronousOperationInfo) -> {
                    if (asynchronousOperationInfo.getFailureCause() == null) {
                        return Acknowledge.get();
                    } else {
                        throw new CompletionException(asynchronousOperationInfo.getFailureCause());
                    }
                });
    }
@Test
    void testDisposeSavepoint() throws Exception {
        final String savepointPath = "foobar";
        final String exceptionMessage = "Test exception.";
        final FlinkException testException = new FlinkException(exceptionMessage);

        final TestSavepointDisposalHandlers testSavepointDisposalHandlers =
                new TestSavepointDisposalHandlers(savepointPath);
        final TestSavepointDisposalHandlers.TestSavepointDisposalTriggerHandler
                testSavepointDisposalTriggerHandler =
                        testSavepointDisposalHandlers.new TestSavepointDisposalTriggerHandler();
        final TestSavepointDisposalHandlers.TestSavepointDisposalStatusHandler
                testSavepointDisposalStatusHandler =
                        testSavepointDisposalHandlers
                        .new TestSavepointDisposalStatusHandler(
                                OptionalFailure.of(AsynchronousOperationInfo.complete()),
                                OptionalFailure.of(
                                        AsynchronousOperationInfo.completeExceptional(
                                                new SerializedThrowable(testException))),
                                OptionalFailure.ofFailure(testException));

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

            try {
                {
                    final CompletableFuture<Acknowledge> disposeSavepointFuture =
                            restClusterClient.disposeSavepoint(savepointPath);
                    assertThat(disposeSavepointFuture.get()).isEqualTo(Acknowledge.get());
                }

                {
                    final CompletableFuture<Acknowledge> disposeSavepointFuture =
                            restClusterClient.disposeSavepoint(savepointPath);

                    try {
                        disposeSavepointFuture.get();
                        fail("Expected an exception");
                    } catch (ExecutionException ee) {
                        assertThat(ExceptionUtils.findThrowableWithMessage(ee, exceptionMessage))
                                .isPresent();
                    }
                }

                {
                    try {
                        restClusterClient.disposeSavepoint(savepointPath).get();
                        fail("Expected an exception.");
                    } catch (ExecutionException ee) {
                        assertThat(ExceptionUtils.findThrowable(ee, RestClientException.class))
                                .isPresent();
                    }
                }
            } finally {
                restClusterClient.close();
            }
        }
    }