class SubtasksAllAccumulatorsHandler extends AbstractJobVertexRequestHandler {

	public static String createSubtasksAccumulatorsJson(AccessExecutionJobVertex jobVertex) throws IOException {
		StringWriter writer = new StringWriter();
		JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);

		gen.writeStartObject();
		gen.writeStringField("id", jobVertex.getJobVertexId().toString());
		gen.writeNumberField("parallelism", jobVertex.getParallelism());

		gen.writeArrayFieldStart("subtasks");

		int num = 0;
		for (AccessExecutionVertex vertex : jobVertex.getTaskVertices()) {

			TaskManagerLocation location = vertex.getCurrentAssignedResourceLocation();
			String locationString = location == null ? "(unassigned)" : location.getHostname();

			gen.writeStartObject();

			gen.writeNumberField("subtask", num++);
			gen.writeNumberField("attempt", vertex.getCurrentExecutionAttempt().getAttemptNumber());
			gen.writeStringField("host", locationString);

			StringifiedAccumulatorResult[] accs = vertex.getCurrentExecutionAttempt().getUserAccumulatorsStringified();
			gen.writeArrayFieldStart("user-accumulators");
			for (StringifiedAccumulatorResult acc : accs) {
				gen.writeStartObject();
				gen.writeStringField("name", acc.getName());
				gen.writeStringField("type", acc.getType());
				gen.writeStringField("value", acc.getValue());
				gen.writeEndObject();
			}
			gen.writeEndArray();

			gen.writeEndObject();
		}
		gen.writeEndArray();

		gen.writeEndObject();
		gen.close();
		return writer.toString();
	}

	public  SubtasksAllAccumulatorsHandler(ExecutionGraphCache executionGraphHolder, Executor executor);

}