class JobCheckpointsHandler extends AbstractExecutionGraphRequestHandler {

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

		CheckpointStatsTracker tracker = graph.getCheckpointStatsTracker();

		gen.writeStartObject();

		if (tracker != null) {
			Option<JobCheckpointStats> stats = tracker.getJobStats();

			if (stats.isDefined()) {
				// Total number of checkpoints
				gen.writeNumberField("count", stats.get().getCount());

				// Duration
				gen.writeFieldName("duration");
				gen.writeStartObject();
				gen.writeNumberField("min", stats.get().getMinDuration());
				gen.writeNumberField("max", stats.get().getMaxDuration());
				gen.writeNumberField("avg", stats.get().getAverageDuration());
				gen.writeEndObject();

				// State size
				gen.writeFieldName("size");
				gen.writeStartObject();
				gen.writeNumberField("min", stats.get().getMinStateSize());
				gen.writeNumberField("max", stats.get().getMaxStateSize());
				gen.writeNumberField("avg", stats.get().getAverageStateSize());
				gen.writeEndObject();

				// Recent history
				gen.writeArrayFieldStart("history");
				for (CheckpointStats checkpoint : stats.get().getRecentHistory()) {
					gen.writeStartObject();
					gen.writeNumberField("id", checkpoint.getCheckpointId());
					gen.writeNumberField("timestamp", checkpoint.getTriggerTimestamp());
					gen.writeNumberField("duration", checkpoint.getDuration());
					gen.writeNumberField("size", checkpoint.getStateSize());
					gen.writeEndObject();
				}
				gen.writeEndArray();
			}
		}

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

		return writer.toString();
	}

	public  JobCheckpointsHandler(ExecutionGraphHolder executionGraphHolder);

}