@Override
    public CompletableFuture<Void> invalidateClusterDataset(AbstractID clusterDatasetId) {

        final ClusterDataSetDeleteTriggerHeaders triggerHeader =
                ClusterDataSetDeleteTriggerHeaders.INSTANCE;
        final ClusterDataSetDeleteTriggerMessageParameters parameters =
                triggerHeader.getUnresolvedMessageParameters();
        parameters.clusterDataSetIdPathParameter.resolve(
                new IntermediateDataSetID(clusterDatasetId));

        final CompletableFuture<TriggerResponse> triggerFuture =
                sendRequest(triggerHeader, parameters);

        final CompletableFuture<AsynchronousOperationInfo> clusterDatasetDeleteFuture =
                triggerFuture.thenCompose(
                        triggerResponse -> {
                            final TriggerId triggerId = triggerResponse.getTriggerId();
                            final ClusterDataSetDeleteStatusHeaders statusHeaders =
                                    ClusterDataSetDeleteStatusHeaders.INSTANCE;
                            final ClusterDataSetDeleteStatusMessageParameters
                                    statusMessageParameters =
                                            statusHeaders.getUnresolvedMessageParameters();
                            statusMessageParameters.triggerIdPathParameter.resolve(triggerId);

                            return pollResourceAsync(
                                    () -> sendRequest(statusHeaders, statusMessageParameters));
                        });

        return clusterDatasetDeleteFuture.thenApply(
                asynchronousOperationInfo -> {
                    if (asynchronousOperationInfo.getFailureCause() == null) {
                        return null;
                    } else {
                        throw new CompletionException(asynchronousOperationInfo.getFailureCause());
                    }
                });
    }
@Test
    public void testInvalidateClusterDataset() throws Exception {
        final IntermediateDataSetID intermediateDataSetID = new IntermediateDataSetID();
        final String exceptionMessage = "Test exception.";
        final FlinkException testException = new FlinkException(exceptionMessage);

        final TestClusterDatasetDeleteHandlers testClusterDatasetDeleteHandlers =
                new TestClusterDatasetDeleteHandlers(intermediateDataSetID);
        final TestClusterDatasetDeleteHandlers.TestClusterDatasetDeleteTriggerHandler
                testClusterDatasetDeleteTriggerHandler =
                        testClusterDatasetDeleteHandlers
                        .new TestClusterDatasetDeleteTriggerHandler();
        final TestClusterDatasetDeleteHandlers.TestClusterDatasetDeleteStatusHandler
                testClusterDatasetDeleteStatusHandler =
                        testClusterDatasetDeleteHandlers
                        .new TestClusterDatasetDeleteStatusHandler(
                                OptionalFailure.of(AsynchronousOperationInfo.complete()),
                                OptionalFailure.of(
                                        AsynchronousOperationInfo.completeExceptional(
                                                new SerializedThrowable(testException))),
                                OptionalFailure.ofFailure(testException));

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

            try {
                {
                    final CompletableFuture<Void> invalidateCacheFuture =
                            restClusterClient.invalidateClusterDataset(intermediateDataSetID);
                    assertThat(invalidateCacheFuture.get()).isNull();
                }

                {
                    final CompletableFuture<Void> invalidateCacheFuture =
                            restClusterClient.invalidateClusterDataset(intermediateDataSetID);

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

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