class JobVertexBackPressureHandler extends AbstractRestHandler<RestfulGateway, EmptyRequestBody, JobVertexBackPressureInfo, JobVertexMessageParameters> {

	@Override
	protected CompletableFuture<JobVertexBackPressureInfo> handleRequest(
			@Nonnull HandlerRequest<EmptyRequestBody, JobVertexMessageParameters> request,
			@Nonnull RestfulGateway gateway) throws RestHandlerException {
		final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
		final JobVertexID jobVertexId = request.getPathParameter(JobVertexIdPathParameter.class);
		return gateway
			.requestOperatorBackPressureStats(jobId, jobVertexId)
			.thenApply(
				operatorBackPressureStats ->
					operatorBackPressureStats.getOperatorBackPressureStats().map(
						JobVertexBackPressureHandler::createJobVertexBackPressureInfo).orElse(
						JobVertexBackPressureInfo.deprecated()));
	}

	public  JobVertexBackPressureHandler(
			GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			Time timeout,
			Map<String, String> responseHeaders,
			MessageHeaders<EmptyRequestBody, JobVertexBackPressureInfo, JobVertexMessageParameters> messageHeaders);

}