@Test public void testHandleRequest() throws Exception { final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(new Configuration()); final SubtaskExecutionAttemptAccumulatorsHandler handler = new SubtaskExecutionAttemptAccumulatorsHandler( () -> null, Time.milliseconds(100L), Collections.emptyMap(), SubtaskExecutionAttemptAccumulatorsHeaders.getInstance(), new ExecutionGraphCache( restHandlerConfiguration.getTimeout(), Time.milliseconds(restHandlerConfiguration.getRefreshInterval())), TestingUtils.defaultExecutor()); final HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request = new HandlerRequest<>( EmptyRequestBody.getInstance(), new SubtaskAttemptMessageParameters() ); final Map<String, OptionalFailure<Accumulator<?, ?>>> userAccumulators = new HashMap<>(3); userAccumulators.put("IntCounter", OptionalFailure.of(new IntCounter(10))); userAccumulators.put("LongCounter", OptionalFailure.of(new LongCounter(100L))); userAccumulators.put("Failure", OptionalFailure.ofFailure(new FlinkRuntimeException("Test"))); final StringifiedAccumulatorResult[] accumulatorResults = StringifiedAccumulatorResult.stringifyAccumulatorResults(userAccumulators); final int attemptNum = 1; final int subtaskIndex = 2; final ArchivedExecution execution = new ArchivedExecution( accumulatorResults, null, new ExecutionAttemptID(), attemptNum, ExecutionState.FINISHED, null, null, null, subtaskIndex, new long[ExecutionState.values().length]); final SubtaskExecutionAttemptAccumulatorsInfo accumulatorsInfo = handler.handleRequest(request, execution); final ArrayList<UserAccumulator> userAccumulatorList = new ArrayList<>(userAccumulators.size()); for (StringifiedAccumulatorResult accumulatorResult : accumulatorResults) { userAccumulatorList.add( new UserAccumulator( accumulatorResult.getName(), accumulatorResult.getType(), accumulatorResult.getValue())); } final SubtaskExecutionAttemptAccumulatorsInfo expected = new SubtaskExecutionAttemptAccumulatorsInfo( subtaskIndex, attemptNum, execution.getAttemptId().toString(), userAccumulatorList); assertEquals(expected, accumulatorsInfo); }
@Override protected SubtaskExecutionAttemptAccumulatorsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { return createAccumulatorInfo(execution); }
SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptAccumulatorsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { return createAccumulatorInfo(execution); } }
SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptAccumulatorsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { return createAccumulatorInfo(execution); } SubtaskExecutionAttemptAccumulatorsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor); }
SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptAccumulatorsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { return createAccumulatorInfo(execution); } SubtaskExecutionAttemptAccumulatorsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor); @Override Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph); }
SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptAccumulatorsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { return createAccumulatorInfo(execution); } SubtaskExecutionAttemptAccumulatorsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptAccumulatorsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor); @Override Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph);  }