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(); } JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder); @Override String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params);  }
@Test public void testNoCoordinator() throws Exception { JobVertexCheckpointsHandler handler = new JobVertexCheckpointsHandler( mock(ExecutionGraphHolder.class)); ExecutionGraph graph = mock(ExecutionGraph.class); ExecutionJobVertex vertex = mock(ExecutionJobVertex.class); when(vertex.getGraph()).thenReturn(graph); String response = handler.handleRequest(vertex, Collections.<String, String>emptyMap()); assertEquals("{}", response); }
@Test public void testNoStats() throws Exception { JobVertexCheckpointsHandler handler = new JobVertexCheckpointsHandler( mock(ExecutionGraphHolder.class)); ExecutionGraph graph = mock(ExecutionGraph.class); ExecutionJobVertex vertex = mock(ExecutionJobVertex.class); CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class); when(vertex.getGraph()).thenReturn(graph); when(graph.getCheckpointStatsTracker()).thenReturn(tracker); when(tracker.getOperatorStats(any(JobVertexID.class))) .thenReturn(Option.<OperatorCheckpointStats>empty()); String response = handler.handleRequest(vertex, Collections.<String, String>emptyMap()); assertEquals("{}", response); }
@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 } }; OperatorCheckpointStats stats = new OperatorCheckpointStats( 3, 6812, 2800, 1024, subTaskStats); when(tracker.getOperatorStats(eq(vertexId))) .thenReturn(Option.apply(stats)); String response = handler.handleRequest(vertex, Collections.<String, String>emptyMap()); ObjectMapper mapper = new ObjectMapper(); JsonNode rootNode = mapper.readTree(response); 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); 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()); }