class JobVertexWatermarksHandler extends AbstractJobVertexHandler<MetricCollectionResponseBody, JobVertexMessageParameters> {

	@Override
	protected MetricCollectionResponseBody handleRequest(
			HandlerRequest<EmptyRequestBody, JobVertexMessageParameters> 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);
	}

	public  JobVertexWatermarksHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MetricFetcher metricFetcher,
			ExecutionGraphCache executionGraphCache,
			Executor executor);

}