class JobVertexCheckpointsHandler extends AbstractJobVertexRequestHandler {

	@Override
	public String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params) throws Exception {
		StringWriter writer = new StringWriter();
		JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
		gen.writeStartObject();

		CheckpointStatsTracker tracker = jobVertex.getGraph().getCheckpointStatsTracker();

		if (tracker != null) {
			Option<OperatorCheckpointStats> statsOption = tracker
					.getOperatorStats(jobVertex.getJobVertexId());

			if (statsOption.isDefined()) {
				OperatorCheckpointStats stats = statsOption.get();

				gen.writeNumberField("id", stats.getCheckpointId());
				gen.writeNumberField("timestamp", stats.getTriggerTimestamp());
				gen.writeNumberField("duration", stats.getDuration());
				gen.writeNumberField("size", stats.getStateSize());
				gen.writeNumberField("parallelism", stats.getNumberOfSubTasks());

				gen.writeArrayFieldStart("subtasks");
				for (int i = 0; i < stats.getNumberOfSubTasks(); i++) {
					gen.writeStartObject();
					gen.writeNumberField("subtask", i);
					gen.writeNumberField("duration", stats.getSubTaskDuration(i));
					gen.writeNumberField("size", stats.getSubTaskStateSize(i));
					gen.writeEndObject();
				}
				gen.writeEndArray();
			}
		}

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

		return writer.toString();
	}

	public  JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder);

}