@Test public void testHandleRequest() throws Exception { final JobID jobID = new JobID(); final JobVertexID jobVertexID = new JobVertexID(); final long deployingTs = System.currentTimeMillis() - 1024; final long finishedTs = System.currentTimeMillis(); final long bytesIn = 1L; final long bytesOut = 10L; final long recordsIn = 20L; final long recordsOut = 30L; final IOMetrics ioMetrics = new IOMetrics( bytesIn, bytesOut, recordsIn, recordsOut); final long[] timestamps = new long[ExecutionState.values().length]; timestamps[ExecutionState.DEPLOYING.ordinal()] = deployingTs; final ExecutionState expectedState = ExecutionState.FINISHED; timestamps[expectedState.ordinal()] = finishedTs; final LocalTaskManagerLocation assignedResourceLocation = new LocalTaskManagerLocation(); final AllocationID allocationID = new AllocationID(); final int subtaskIndex = 1; final int attempt = 2; final ArchivedExecution execution = new ArchivedExecution( new StringifiedAccumulatorResult[0], ioMetrics, new ExecutionAttemptID(), attempt, expectedState, null, assignedResourceLocation, allocationID, subtaskIndex, timestamps); final ArchivedExecutionVertex executionVertex = new ArchivedExecutionVertex( subtaskIndex, "Test archived execution vertex", execution, new EvictingBoundedList<>(0)); final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(new Configuration()); final MetricFetcher metricFetcher = new MetricFetcherImpl<>( () -> null, address -> null, TestingUtils.defaultExecutor(), Time.milliseconds(1000L), MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue()); final SubtaskCurrentAttemptDetailsHandler handler = new SubtaskCurrentAttemptDetailsHandler( () -> null, Time.milliseconds(100), Collections.emptyMap(), SubtaskCurrentAttemptDetailsHeaders.getInstance(), new ExecutionGraphCache( restHandlerConfiguration.getTimeout(), Time.milliseconds(restHandlerConfiguration.getRefreshInterval())), TestingUtils.defaultExecutor(), metricFetcher); final HashMap<String, String> receivedPathParameters = new HashMap<>(2); receivedPathParameters.put(JobIDPathParameter.KEY, jobID.toString()); receivedPathParameters.put(JobVertexIdPathParameter.KEY, jobVertexID.toString()); final HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request = new HandlerRequest<>( EmptyRequestBody.getInstance(), new SubtaskMessageParameters(), receivedPathParameters, Collections.emptyMap()); final SubtaskExecutionAttemptDetailsInfo detailsInfo = handler.handleRequest(request, executionVertex); final IOMetricsInfo ioMetricsInfo = new IOMetricsInfo( bytesIn, true, bytesOut, true, recordsIn, true, recordsOut, true ); final SubtaskExecutionAttemptDetailsInfo expectedDetailsInfo = new SubtaskExecutionAttemptDetailsInfo( subtaskIndex, expectedState, attempt, assignedResourceLocation.getHostname(), deployingTs, finishedTs, finishedTs - deployingTs, ioMetricsInfo ); assertEquals(expectedDetailsInfo, detailsInfo); }
@Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request, AccessExecutionVertex executionVertex) throws RestHandlerException { final AccessExecution execution = executionVertex.getCurrentExecutionAttempt(); final MutableIOMetrics ioMetrics = new MutableIOMetrics(); final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); ioMetrics.addIOMetrics( execution, metricFetcher, jobID.toString(), jobVertexID.toString() ); return SubtaskExecutionAttemptDetailsInfo.create(execution, ioMetrics); }
SubtaskCurrentAttemptDetailsHandler extends AbstractSubtaskHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request, AccessExecutionVertex executionVertex) throws RestHandlerException { final AccessExecution execution = executionVertex.getCurrentExecutionAttempt(); final MutableIOMetrics ioMetrics = new MutableIOMetrics(); final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); ioMetrics.addIOMetrics( execution, metricFetcher, jobID.toString(), jobVertexID.toString() ); return SubtaskExecutionAttemptDetailsInfo.create(execution, ioMetrics); } }
SubtaskCurrentAttemptDetailsHandler extends AbstractSubtaskHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request, AccessExecutionVertex executionVertex) throws RestHandlerException { final AccessExecution execution = executionVertex.getCurrentExecutionAttempt(); final MutableIOMetrics ioMetrics = new MutableIOMetrics(); final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); ioMetrics.addIOMetrics( execution, metricFetcher, jobID.toString(), jobVertexID.toString() ); return SubtaskExecutionAttemptDetailsInfo.create(execution, ioMetrics); } SubtaskCurrentAttemptDetailsHandler(
		GatewayRetriever<? extends RestfulGateway> leaderRetriever,
		Time timeout,
		Map<String, String> responseHeaders,
		MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> messageHeaders,
		ExecutionGraphCache executionGraphCache,
		Executor executor,
		MetricFetcher metricFetcher); }
SubtaskCurrentAttemptDetailsHandler extends AbstractSubtaskHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request, AccessExecutionVertex executionVertex) throws RestHandlerException { final AccessExecution execution = executionVertex.getCurrentExecutionAttempt(); final MutableIOMetrics ioMetrics = new MutableIOMetrics(); final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); ioMetrics.addIOMetrics( execution, metricFetcher, jobID.toString(), jobVertexID.toString() ); return SubtaskExecutionAttemptDetailsInfo.create(execution, ioMetrics); } SubtaskCurrentAttemptDetailsHandler(
		GatewayRetriever<? extends RestfulGateway> leaderRetriever,
		Time timeout,
		Map<String, String> responseHeaders,
		MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> messageHeaders,
		ExecutionGraphCache executionGraphCache,
		Executor executor,
		MetricFetcher metricFetcher);  }
SubtaskCurrentAttemptDetailsHandler extends AbstractSubtaskHandler<SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> { @Override protected SubtaskExecutionAttemptDetailsInfo handleRequest( HandlerRequest<EmptyRequestBody, SubtaskMessageParameters> request, AccessExecutionVertex executionVertex) throws RestHandlerException { final AccessExecution execution = executionVertex.getCurrentExecutionAttempt(); final MutableIOMetrics ioMetrics = new MutableIOMetrics(); final JobID jobID = request.getPathParameter(JobIDPathParameter.class); final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class); ioMetrics.addIOMetrics( execution, metricFetcher, jobID.toString(), jobVertexID.toString() ); return SubtaskExecutionAttemptDetailsInfo.create(execution, ioMetrics); } SubtaskCurrentAttemptDetailsHandler(
		GatewayRetriever<? extends RestfulGateway> leaderRetriever,
		Time timeout,
		Map<String, String> responseHeaders,
		MessageHeaders<EmptyRequestBody, SubtaskExecutionAttemptDetailsInfo, SubtaskMessageParameters> messageHeaders,
		ExecutionGraphCache executionGraphCache,
		Executor executor,
		MetricFetcher metricFetcher);   }