@Override
    protected MetricCollectionResponseBody handleRequest(
            HandlerRequest<EmptyRequestBody> request, AccessExecutionJobVertex jobVertex)
            throws RestHandlerException {

        String jobID = request.getPathParameter(JobIDPathParameter.class).toString();
        String taskID = jobVertex.getJobVertexId().toString();

        metricFetcher.update();
        MetricStore.TaskMetricStore taskMetricStore =
                metricFetcher.getMetricStore().getTaskMetricStore(jobID, taskID);
        if (taskMetricStore == null) {
            return new MetricCollectionResponseBody(Collections.emptyList());
        }

        AccessExecutionVertex[] taskVertices = jobVertex.getTaskVertices();
        List<Metric> metrics = new ArrayList<>(taskVertices.length);

        for (AccessExecutionVertex taskVertex : taskVertices) {
            String id =
                    taskVertex.getParallelSubtaskIndex()
                            + "."
                            + MetricNames.IO_CURRENT_INPUT_WATERMARK;
            String watermarkValue = taskMetricStore.getMetric(id);
            if (watermarkValue != null) {
                metrics.add(new Metric(id, watermarkValue));
            }
        }

        return new MetricCollectionResponseBody(metrics);
    }
@Test
    void testWatermarksRetrieval() throws Exception {
        Mockito.when(taskMetricStore.getMetric("0.currentInputWatermark")).thenReturn("23");
        Mockito.when(taskMetricStore.getMetric("1.currentInputWatermark")).thenReturn("42");

        MetricCollectionResponseBody response = watermarkHandler.handleRequest(request, vertex);

        assertThat(response.getMetrics())
                .satisfies(
                        matching(
                                containsInAnyOrder(
                                        new MetricMatcher("0.currentInputWatermark", "23"),
                                        new MetricMatcher("1.currentInputWatermark", "42"))));
    }