class RestClusterClient extends ClusterClient<T> implements NewClusterClient {

	@Override
	public Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader) throws Exception {
		final JobAccumulatorsHeaders accumulatorsHeaders = JobAccumulatorsHeaders.getInstance();
		final JobAccumulatorsMessageParameters accMsgParams = accumulatorsHeaders.getUnresolvedMessageParameters();
		accMsgParams.jobPathParameter.resolve(jobID);
		accMsgParams.includeSerializedAccumulatorsParameter.resolve(Collections.singletonList(true));

		CompletableFuture<JobAccumulatorsInfo> responseFuture = sendRequest(
			accumulatorsHeaders,
			accMsgParams
		);

		Map<String, OptionalFailure<Object>> result = Collections.emptyMap();

		try {
			result = responseFuture.thenApply((JobAccumulatorsInfo accumulatorsInfo) -> {
				try {
					return AccumulatorHelper.deserializeAccumulators(
						accumulatorsInfo.getSerializedUserAccumulators(),
						loader);
				} catch (Exception e) {
					throw new CompletionException(
						new FlinkException(
							String.format("Deserialization of accumulators for job %s failed.", jobID),
							e));
				}
			}).get(timeout.toMillis(), TimeUnit.MILLISECONDS);
		} catch (ExecutionException ee) {
			ExceptionUtils.rethrowException(ExceptionUtils.stripExecutionException(ee));
		}

		return result;
	}

	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);

}