@Override
    protected CompletableFuture<JobVertexBackPressureInfo> handleRequest(
            @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
            throws RestHandlerException {
        metricFetcher.update();

        final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
        final JobVertexID jobVertexId = request.getPathParameter(JobVertexIdPathParameter.class);

        TaskMetricStore taskMetricStore =
                metricFetcher
                        .getMetricStore()
                        .getTaskMetricStore(jobId.toString(), jobVertexId.toString());
        Map<String, Map<Integer, Integer>> jobRepresentativeExecutions =
                metricFetcher.getMetricStore().getRepresentativeAttempts().get(jobId.toString());
        Map<Integer, Integer> representativeAttempts =
                jobRepresentativeExecutions != null
                        ? jobRepresentativeExecutions.get(jobVertexId.toString())
                        : null;

        return CompletableFuture.completedFuture(
                taskMetricStore != null
                        ? createJobVertexBackPressureInfo(taskMetricStore, representativeAttempts)
                        : JobVertexBackPressureInfo.deprecated());
    }
@Test
    void testAbsentBackPressure() throws Exception {
        final Map<String, String> pathParameters = new HashMap<>();
        pathParameters.put(
                JobIDPathParameter.KEY, TEST_JOB_ID_BACK_PRESSURE_STATS_ABSENT.toString());
        pathParameters.put(JobVertexIdPathParameter.KEY, new JobVertexID().toString());

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

        final CompletableFuture<JobVertexBackPressureInfo>
                jobVertexBackPressureInfoCompletableFuture =
                        jobVertexBackPressureHandler.handleRequest(request, restfulGateway);
        final JobVertexBackPressureInfo jobVertexBackPressureInfo =
                jobVertexBackPressureInfoCompletableFuture.get();

        assertThat(jobVertexBackPressureInfo.getStatus())
                .isEqualTo(VertexBackPressureStatus.DEPRECATED);
    }