@Test public void testHandleRequest() throws Exception { final JobID jobID = new JobID(); final JobVertexID jobVertexId = new JobVertexID(); final int subtaskIndex = 1; final ExecutionState expectedState = ExecutionState.FINISHED; final int attempt = 0; final StringifiedAccumulatorResult[] emptyAccumulators = new StringifiedAccumulatorResult[0]; final long bytesInLocal = 1L; final long bytesInRemote = 2L; final long bytesOut = 10L; final long recordsIn = 20L; final long recordsOut = 30L; final IOMetrics ioMetrics = new IOMetrics( bytesInLocal, bytesInRemote, bytesOut, recordsIn, recordsOut, 0.0, 0.0, 0.0, 0.0, 0.0); final ArchivedExecutionJobVertex archivedExecutionJobVertex = new ArchivedExecutionJobVertex( new ArchivedExecutionVertex[]{ null, new ArchivedExecutionVertex( subtaskIndex, "test task", new ArchivedExecution( emptyAccumulators, ioMetrics, new ExecutionAttemptID(), attempt, expectedState, null, null, null, subtaskIndex, new long[ExecutionState.values().length]), new EvictingBoundedList<>(0) ) }, jobVertexId, "test", 1, 1, emptyAccumulators); final MetricFetcher metricFetcher = new MetricFetcherImpl<>( () -> null, path -> null, TestingUtils.defaultExecutor(), Time.milliseconds(1000L), MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue()); final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(new Configuration()); final SubtaskExecutionAttemptDetailsHandler handler = new SubtaskExecutionAttemptDetailsHandler( () -> null, Time.milliseconds(100L), Collections.emptyMap(), SubtaskExecutionAttemptDetailsHeaders.getInstance(), new ExecutionGraphCache( restHandlerConfiguration.getTimeout(), Time.milliseconds(restHandlerConfiguration.getRefreshInterval())), TestingUtils.defaultExecutor(), metricFetcher); final HashMap<String, String> receivedPathParameters = new HashMap<>(4); receivedPathParameters.put(JobIDPathParameter.KEY, jobID.toString()); receivedPathParameters.put(JobVertexIdPathParameter.KEY, jobVertexId.toString()); receivedPathParameters.put(SubtaskIndexPathParameter.KEY, Integer.toString(subtaskIndex)); receivedPathParameters.put(SubtaskAttemptPathParameter.KEY, Integer.toString(attempt)); final HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request = new HandlerRequest<>( EmptyRequestBody.getInstance(), new SubtaskAttemptMessageParameters(), receivedPathParameters, Collections.emptyMap() ); final SubtaskExecutionAttemptDetailsInfo detailsInfo = handler.handleRequest( request, archivedExecutionJobVertex); final IOMetricsInfo ioMetricsInfo = new IOMetricsInfo( bytesInLocal + bytesInRemote, true, bytesOut, true, recordsIn, true, recordsOut, true ); final SubtaskExecutionAttemptDetailsInfo expectedDetailsInfo = new SubtaskExecutionAttemptDetailsInfo( subtaskIndex, expectedState, attempt, "(unassigned)", -1L, 0L, -1L, ioMetricsInfo ); assertEquals(expectedDetailsInfo, detailsInfo); }
@Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); return createDetailsInfo(execution, jobID, jobVertexID, metricFetcher); }
SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); return createDetailsInfo(execution, jobID, jobVertexID, metricFetcher); } }
SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); return createDetailsInfo(execution, jobID, jobVertexID, metricFetcher); } SubtaskExecutionAttemptDetailsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor,
			MetricFetcher metricFetcher); }
SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); return createDetailsInfo(execution, jobID, jobVertexID, metricFetcher); } SubtaskExecutionAttemptDetailsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor,
			MetricFetcher metricFetcher); @Override Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph); }
SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> implements JsonArchivist { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request, AccessExecution execution) throws RestHandlerException { final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); return createDetailsInfo(execution, jobID, jobVertexID, metricFetcher); } SubtaskExecutionAttemptDetailsHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskAttemptMessageParameters> messageHeaders,
			ExecutionGraphCache executionGraphCache,
			Executor executor,
			MetricFetcher metricFetcher); @Override Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph);  }