class RestClusterClient extends ClusterClient<T> implements NewClusterClient {

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

		final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest(
			SavepointDisposalTriggerHeaders.getInstance(),
			EmptyMessageParameters.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(
					() -> sendRetryableRequest(
						savepointDisposalStatusHeaders,
						savepointDisposalStatusMessageParameters,
						EmptyRequestBody.getInstance(),
						isConnectionProblemException()));
			});

		return savepointDisposalFuture.thenApply(
			(AsynchronousOperationInfo asynchronousOperationInfo) -> {
				if (asynchronousOperationInfo.getFailureCause() == null) {
					return Acknowledge.get();
				} else {
					throw new CompletionException(asynchronousOperationInfo.getFailureCause());
				}
			});
	}

	public  RestClusterClient(Configuration config, T clusterId);
	public  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService);
	@VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService);

}