test_class.file: flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandlerTest.java
test_class.identifier: JobVertexCheckpointsHandlerTest
test_case.body: @Test
  	public void testStats() throws Exception {
  		JobVertexCheckpointsHandler handler = new JobVertexCheckpointsHandler(
  				mock(ExecutionGraphHolder.class));
  
  		JobVertexID vertexId = new JobVertexID();
  
  		ExecutionGraph graph = mock(ExecutionGraph.class);
  		ExecutionJobVertex vertex = mock(ExecutionJobVertex.class);
  		CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class);
  
  		when(vertex.getJobVertexId()).thenReturn(vertexId);
  		when(vertex.getGraph()).thenReturn(graph);
  		when(graph.getCheckpointStatsTracker()).thenReturn(tracker);
  
  		long[][] subTaskStats = new long[][] {
  				new long[] { 1, 10 },
  				new long[] { 2, 9 },
  				new long[] { 3, 8 },
  				new long[] { 4, 7 },
  				new long[] { 5, 6 },
  				new long[] { 6, 5 },
  				new long[] { 7, 4 },
  				new long[] { 8, 3 },
  				new long[] { 9, 2 },
  				new long[] { 10, 1 } };
  
  		// Stats
  		OperatorCheckpointStats stats = new OperatorCheckpointStats(
  				3, 6812, 2800, 1024, subTaskStats);
  
  		when(tracker.getOperatorStats(eq(vertexId)))
  				.thenReturn(Option.apply(stats));
  
  		// Request stats
  		String response = handler.handleRequest(vertex, Collections.<String, String>emptyMap());
  
  		ObjectMapper mapper = new ObjectMapper();
  		JsonNode rootNode = mapper.readTree(response);
  
  		// Operator stats
  		long checkpointId = rootNode.get("id").asLong();
  		long timestamp = rootNode.get("timestamp").asLong();
  		long duration = rootNode.get("duration").asLong();
  		long size = rootNode.get("size").asLong();
  		long parallelism = rootNode.get("parallelism").asLong();
  
  		assertEquals(stats.getCheckpointId(), checkpointId);
  		assertEquals(stats.getTriggerTimestamp(), timestamp);
  		assertEquals(stats.getDuration(), duration);
  		assertEquals(stats.getStateSize(), size);
  		assertEquals(subTaskStats.length, parallelism);
  
  		// Sub task stats
  		JsonNode subTasksNode = rootNode.get("subtasks");
  		assertNotNull(subTasksNode);
  		assertTrue(subTasksNode.isArray());
  
  		Iterator<JsonNode> it = subTasksNode.elements();
  
  		for (int i = 0; i < subTaskStats.length; i++) {
  			JsonNode node = it.next();
  
  			assertEquals(i, node.get("subtask").asInt());
  			assertEquals(subTaskStats[i][0], node.get("duration").asLong());
  			assertEquals(subTaskStats[i][1], node.get("size").asLong());
  		}
  
  		assertFalse(it.hasNext());
  	}
test_case.class_method_signature: JobVertexCheckpointsHandlerTest.testStats()
test_case.full_signature: @Test public void testStats()
test_case.identifier: testStats
test_case.invocations: mock mock mock mock thenReturn when getJobVertexId thenReturn when getGraph thenReturn when getCheckpointStatsTracker thenReturn when getOperatorStats eq apply handleRequest emptyMap readTree asLong get asLong get asLong get asLong get asLong get assertEquals getCheckpointId assertEquals getTriggerTimestamp assertEquals getDuration assertEquals getStateSize assertEquals get assertNotNull assertTrue isArray elements next assertEquals asInt get assertEquals asLong get assertEquals asLong get assertFalse hasNext
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testStats()
focal_class.file: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java
focal_class.identifier: JobVertexCheckpointsHandler
focal_class.class_method_signature: JobVertexCheckpointsHandler.JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder)
focal_class.full_signature: public  JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder)
focal_class.identifier: JobVertexCheckpointsHandler
focal_class.modifiers: public
focal_class.parameters: (ExecutionGraphHolder executionGraphHolder)
focal_class.signature:  JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder)
focal_class.class_method_signature: JobVertexCheckpointsHandler.handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)
focal_class.full_signature: @Override public String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)
focal_class.identifier: handleRequest
focal_class.modifiers: @Override public
focal_class.parameters: (ExecutionJobVertex jobVertex, Map<String, String> params)
focal_class.return: String
focal_class.signature: String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)
focal_class.superclass: extends AbstractJobVertexRequestHandler
focal_method.body: @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();
  	}
focal_method.class_method_signature: JobVertexCheckpointsHandler.handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)
focal_method.full_signature: @Override public String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)
focal_method.identifier: handleRequest
focal_method.invocations: createGenerator writeStartObject getCheckpointStatsTracker getGraph getOperatorStats getJobVertexId isDefined get writeNumberField getCheckpointId writeNumberField getTriggerTimestamp writeNumberField getDuration writeNumberField getStateSize writeNumberField getNumberOfSubTasks writeArrayFieldStart getNumberOfSubTasks writeStartObject writeNumberField writeNumberField getSubTaskDuration writeNumberField getSubTaskStateSize writeEndObject writeEndArray writeEndObject close toString
focal_method.modifiers: @Override public
focal_method.parameters: (ExecutionJobVertex jobVertex, Map<String, String> params)
focal_method.return: String
focal_method.signature: String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params)