@Override
    protected SubtaskExecutionAttemptDetailsInfo handleRequest(
            HandlerRequest<EmptyRequestBody> request, AccessExecution execution)
            throws RestHandlerException {

        final JobID jobID = request.getPathParameter(JobIDPathParameter.class);
        final JobVertexID jobVertexID = request.getPathParameter(JobVertexIdPathParameter.class);

        return SubtaskExecutionAttemptDetailsInfo.create(
                execution, metricFetcher, jobID, jobVertexID, null);
    }
@Test
    void testHandleRequest() throws Exception {

        final JobID jobID = new JobID();
        final JobVertexID jobVertexId = new JobVertexID();

        // The testing subtask.
        final int subtaskIndex = 1;
        final ExecutionState expectedState = ExecutionState.FINISHED;
        final int attempt = 0;

        final StringifiedAccumulatorResult[] emptyAccumulators =
                new StringifiedAccumulatorResult[0];

        final long bytesIn = 1L;
        final long bytesOut = 10L;
        final long recordsIn = 20L;
        final long recordsOut = 30L;
        final long accumulateIdleTime = 40L;
        final long accumulateBusyTime = 50L;
        final long accumulateBackPressuredTime = 60L;

        final IOMetrics ioMetrics =
                new IOMetrics(
                        bytesIn,
                        bytesOut,
                        recordsIn,
                        recordsOut,
                        accumulateIdleTime,
                        accumulateBusyTime,
                        accumulateBackPressuredTime);

        final ArchivedExecutionJobVertex archivedExecutionJobVertex =
                new ArchivedExecutionJobVertex(
                        new ArchivedExecutionVertex[] {
                            null, // the first subtask won't be queried
                            new ArchivedExecutionVertex(
                                    subtaskIndex,
                                    "test task",
                                    new ArchivedExecution(
                                            emptyAccumulators,
                                            ioMetrics,
                                            createExecutionAttemptId(
                                                    jobVertexId, subtaskIndex, attempt),
                                            expectedState,
                                            null,
                                            null,
                                            null,
                                            new long[ExecutionState.values().length],
                                            new long[ExecutionState.values().length]),
                                    new ExecutionHistory(0))
                        },
                        jobVertexId,
                        "test",
                        1,
                        1,
                        new SlotSharingGroup(),
                        ResourceProfile.UNKNOWN,
                        emptyAccumulators);

        // Change some fields so we can make it different from other sub tasks.
        final MetricFetcher metricFetcher =
                new MetricFetcherImpl<>(
                        () -> null,
                        address -> null,
                        Executors.directExecutor(),
                        Time.milliseconds(1000L),
                        MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue().toMillis());

        // Instance the handler.
        final RestHandlerConfiguration restHandlerConfiguration =
                RestHandlerConfiguration.fromConfiguration(new Configuration());

        final SubtaskExecutionAttemptDetailsHandler handler =
                new SubtaskExecutionAttemptDetailsHandler(
                        () -> null,
                        Time.milliseconds(100L),
                        Collections.emptyMap(),
                        SubtaskExecutionAttemptDetailsHeaders.getInstance(),
                        new DefaultExecutionGraphCache(
                                restHandlerConfiguration.getTimeout(),
                                Time.milliseconds(restHandlerConfiguration.getRefreshInterval())),
                        Executors.directExecutor(),
                        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> request =
                HandlerRequest.resolveParametersAndCreate(
                        EmptyRequestBody.getInstance(),
                        new SubtaskAttemptMessageParameters(),
                        receivedPathParameters,
                        Collections.emptyMap(),
                        Collections.emptyList());

        // Handle request.
        final SubtaskExecutionAttemptDetailsInfo detailsInfo =
                handler.handleRequest(request, archivedExecutionJobVertex);

        // Verify
        final IOMetricsInfo ioMetricsInfo =
                new IOMetricsInfo(
                        bytesIn,
                        true,
                        bytesOut,
                        true,
                        recordsIn,
                        true,
                        recordsOut,
                        true,
                        accumulateBackPressuredTime,
                        accumulateIdleTime,
                        accumulateBusyTime);

        final Map<ExecutionState, Long> statusDuration = new HashMap<>();
        statusDuration.put(ExecutionState.CREATED, -1L);
        statusDuration.put(ExecutionState.SCHEDULED, -1L);
        statusDuration.put(ExecutionState.DEPLOYING, -1L);
        statusDuration.put(ExecutionState.INITIALIZING, -1L);
        statusDuration.put(ExecutionState.RUNNING, -1L);

        final SubtaskExecutionAttemptDetailsInfo expectedDetailsInfo =
                new SubtaskExecutionAttemptDetailsInfo(
                        subtaskIndex,
                        expectedState,
                        attempt,
                        "(unassigned)",
                        "(unassigned)",
                        -1L,
                        0L,
                        -1L,
                        ioMetricsInfo,
                        "(unassigned)",
                        statusDuration,
                        null);

        assertThat(detailsInfo).isEqualTo(expectedDetailsInfo);
    }