class JobVertexBackPressureHandler extends AbstractJobVertexRequestHandler {

	@Override
	public CompletableFuture<String> handleRequest(
			AccessExecutionJobVertex accessJobVertex,
			Map<String, String> params) {
		if (accessJobVertex instanceof ArchivedExecutionJobVertex) {
			return CompletableFuture.completedFuture("");
		}
		ExecutionJobVertex jobVertex = (ExecutionJobVertex) accessJobVertex;
		try (StringWriter writer = new StringWriter();
				JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer)) {

			gen.writeStartObject();

			Optional<OperatorBackPressureStats> statsOption = backPressureStatsTrackerImpl
					.getOperatorBackPressureStats(jobVertex);

			if (statsOption.isPresent()) {
				OperatorBackPressureStats stats = statsOption.get();

				// Check whether we need to refresh
				if (refreshInterval <= System.currentTimeMillis() - stats.getEndTimestamp()) {
					backPressureStatsTrackerImpl.triggerStackTraceSample(jobVertex);
					gen.writeStringField("status", "deprecated");
				} else {
					gen.writeStringField("status", "ok");
				}

				gen.writeStringField("backpressure-level", getBackPressureLevel(stats.getMaxBackPressureRatio()));
				gen.writeNumberField("end-timestamp", stats.getEndTimestamp());

				// Sub tasks
				gen.writeArrayFieldStart("subtasks");
				int numSubTasks = stats.getNumberOfSubTasks();
				for (int i = 0; i < numSubTasks; i++) {
					double ratio = stats.getBackPressureRatio(i);

					gen.writeStartObject();
					gen.writeNumberField("subtask", i);
					gen.writeStringField("backpressure-level", getBackPressureLevel(ratio));
					gen.writeNumberField("ratio", ratio);
					gen.writeEndObject();
				}
				gen.writeEndArray();
			} else {
				backPressureStatsTrackerImpl.triggerStackTraceSample(jobVertex);
				gen.writeStringField("status", "deprecated");
			}

			gen.writeEndObject();
			gen.close();

			return CompletableFuture.completedFuture(writer.toString());
		} catch (IOException e) {
			return FutureUtils.completedExceptionally(e);
		}
	}

	public  JobVertexBackPressureHandler(
			ExecutionGraphCache executionGraphHolder,
			Executor executor,
			BackPressureStatsTrackerImpl backPressureStatsTrackerImpl,
			int refreshInterval);

}