@Test public void testGetAccumulators() throws Exception { TestAccumulatorHandler accumulatorHandler = new TestAccumulatorHandler(); try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(accumulatorHandler)){ RestClusterClient<?> restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort()); try { JobID id = new JobID(); { Map<String, OptionalFailure<Object>> accumulators = restClusterClient.getAccumulators(id); assertNotNull(accumulators); assertEquals(1, accumulators.size()); assertEquals(true, accumulators.containsKey("testKey")); assertEquals("testValue", accumulators.get("testKey").get().toString()); } } finally { restClusterClient.shutdown(); } } }
@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; }
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; } }
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; } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); }
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; } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); @Override void shutdown(); @Override JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<JobDetailsInfo> getJobDetails(JobID jobId); @Override CompletableFuture<JobStatus> getJobStatus(JobID jobId); @Override CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId); @Override CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph); @Override void cancel(JobID jobID); @Override String stopWithSavepoint(
			final JobID jobId,
			final boolean advanceToEndOfTime,
			@Nullable final String savepointDirectory); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final @Nullable String savepointDirectory); @Override Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override T getClusterId(); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); @Override void shutDownCluster(); @Override String getWebInterfaceURL(); @VisibleForTesting CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request); }
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; } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); @Override void shutdown(); @Override JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<JobDetailsInfo> getJobDetails(JobID jobId); @Override CompletableFuture<JobStatus> getJobStatus(JobID jobId); @Override CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId); @Override CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph); @Override void cancel(JobID jobID); @Override String stopWithSavepoint(
			final JobID jobId,
			final boolean advanceToEndOfTime,
			@Nullable final String savepointDirectory); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final @Nullable String savepointDirectory); @Override Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override T getClusterId(); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); @Override void shutDownCluster(); @Override String getWebInterfaceURL(); @VisibleForTesting CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request);  }