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

        final AccessExecution execution = executionVertex.getCurrentExecutionAttempt();

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

        final Collection<AccessExecution> attempts = executionVertex.getCurrentExecutions();
        List<SubtaskExecutionAttemptDetailsInfo> otherConcurrentAttempts = null;

        if (attempts.size() > 1) {
            otherConcurrentAttempts = new ArrayList<>();
            for (AccessExecution attempt : attempts) {
                if (attempt.getAttemptNumber() != execution.getAttemptNumber()) {
                    otherConcurrentAttempts.add(
                            SubtaskExecutionAttemptDetailsInfo.create(
                                    attempt, metricFetcher, jobID, jobVertexID, null));
                }
            }
        }

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

        // Prepare the execution graph.
        final JobID jobID = new JobID();
        final JobVertexID jobVertexID = new JobVertexID();

        // The testing subtask.
        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 long accumulateIdleTime = 40L;
        final long accumulateBusyTime = 50L;
        final long accumulateBackPressuredTime = 60L;

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

        final long[] timestamps = new long[ExecutionState.values().length];
        final long[] endTimestamps = new long[ExecutionState.values().length];
        timestamps[ExecutionState.DEPLOYING.ordinal()] = deployingTs;
        endTimestamps[ExecutionState.DEPLOYING.ordinal()] = deployingTs + 10;
        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,
                        createExecutionAttemptId(jobVertexID, subtaskIndex, attempt),
                        expectedState,
                        null,
                        assignedResourceLocation,
                        allocationID,
                        timestamps,
                        endTimestamps);

        final ArchivedExecutionVertex executionVertex =
                new ArchivedExecutionVertex(
                        subtaskIndex,
                        "Test archived execution vertex",
                        execution,
                        new ExecutionHistory(0));

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

        final MetricFetcher metricFetcher =
                new MetricFetcherImpl<>(
                        () -> null,
                        address -> null,
                        Executors.directExecutor(),
                        Time.milliseconds(1000L),
                        MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue().toMillis());

        final SubtaskCurrentAttemptDetailsHandler handler =
                new SubtaskCurrentAttemptDetailsHandler(
                        () -> null,
                        Time.milliseconds(100),
                        Collections.emptyMap(),
                        SubtaskCurrentAttemptDetailsHeaders.getInstance(),
                        new DefaultExecutionGraphCache(
                                restHandlerConfiguration.getTimeout(),
                                Time.milliseconds(restHandlerConfiguration.getRefreshInterval())),
                        Executors.directExecutor(),
                        metricFetcher);

        final HashMap<String, String> receivedPathParameters = new HashMap<>(2);
        receivedPathParameters.put(JobIDPathParameter.KEY, jobID.toString());
        receivedPathParameters.put(JobVertexIdPathParameter.KEY, jobVertexID.toString());

        final HandlerRequest<EmptyRequestBody> request =
                HandlerRequest.resolveParametersAndCreate(
                        EmptyRequestBody.getInstance(),
                        new SubtaskMessageParameters(),
                        receivedPathParameters,
                        Collections.emptyMap(),
                        Collections.emptyList());

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

        // 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, 10L);
        statusDuration.put(ExecutionState.INITIALIZING, -1L);
        statusDuration.put(ExecutionState.RUNNING, -1L);

        final SubtaskExecutionAttemptDetailsInfo expectedDetailsInfo =
                new SubtaskExecutionAttemptDetailsInfo(
                        subtaskIndex,
                        expectedState,
                        attempt,
                        assignedResourceLocation.getHostname(),
                        assignedResourceLocation.getEndpoint(),
                        deployingTs,
                        finishedTs,
                        finishedTs - deployingTs,
                        ioMetricsInfo,
                        assignedResourceLocation.getResourceID().getResourceIdString(),
                        statusDuration,
                        null);

        assertThat(detailsInfo).isEqualTo(expectedDetailsInfo);
    }