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()) { gen.writeNumberField("count", stats.get().getCount()); 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(); 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(); 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(); } JobCheckpointsHandler(ExecutionGraphHolder executionGraphHolder); @Override String handleRequest(ExecutionGraph graph, Map<String, String> params);  }
@Test public void testNoCoordinator() throws Exception { JobCheckpointsHandler handler = new JobCheckpointsHandler( mock(ExecutionGraphHolder.class)); ExecutionGraph graph = mock(ExecutionGraph.class); when(graph.getCheckpointStatsTracker()).thenReturn(null); String response = handler.handleRequest(graph, Collections.<String, String>emptyMap()); assertEquals("{}", response); }
@Test public void testNoStats() throws Exception { JobCheckpointsHandler handler = new JobCheckpointsHandler( mock(ExecutionGraphHolder.class)); ExecutionGraph graph = mock(ExecutionGraph.class); CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class); when(graph.getCheckpointStatsTracker()).thenReturn(tracker); when(tracker.getJobStats()).thenReturn(Option.<JobCheckpointStats>empty()); String response = handler.handleRequest(graph, Collections.<String, String>emptyMap()); assertEquals("{}", response); }
@Test public void testStats() throws Exception { JobCheckpointsHandler handler = new JobCheckpointsHandler( mock(ExecutionGraphHolder.class)); ExecutionGraph graph = mock(ExecutionGraph.class); CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class); when(graph.getCheckpointStatsTracker()).thenReturn(tracker); final List<CheckpointStats> history = new ArrayList<>(); history.add(new CheckpointStats(0, 1, 1, 124)); history.add(new CheckpointStats(1, 5, 177, 0)); history.add(new CheckpointStats(2, 6, 8282, 2)); history.add(new CheckpointStats(3, 6812, 2800, 1024)); JobCheckpointStats stats = new JobCheckpointStats() { @Override public List<CheckpointStats> getRecentHistory() { return history; } @Override public long getCount() { return 4; } @Override public long getMinDuration() { return 1; } @Override public long getMaxDuration() { return 8282; } @Override public long getAverageDuration() { return 2815; } @Override public long getMinStateSize() { return 0; } @Override public long getMaxStateSize() { return 1024; } @Override public long getAverageStateSize() { return 287; } }; when(tracker.getJobStats()).thenReturn(Option.apply(stats)); String response = handler.handleRequest(graph, Collections.<String, String>emptyMap()); ObjectMapper mapper = new ObjectMapper(); JsonNode rootNode = mapper.readTree(response); int count = rootNode.get("count").asInt(); assertEquals(stats.getCount(), count); JsonNode durationNode = rootNode.get("duration"); assertNotNull(durationNode); long minDuration = durationNode.get("min").asLong(); long maxDuration = durationNode.get("max").asLong(); long avgDuration = durationNode.get("avg").asLong(); assertEquals(stats.getMinDuration(), minDuration); assertEquals(stats.getMaxDuration(), maxDuration); assertEquals(stats.getAverageDuration(), avgDuration); JsonNode sizeNode = rootNode.get("size"); assertNotNull(sizeNode); long minSize = sizeNode.get("min").asLong(); long maxSize = sizeNode.get("max").asLong(); long avgSize = sizeNode.get("avg").asLong(); assertEquals(stats.getMinStateSize(), minSize); assertEquals(stats.getMaxStateSize(), maxSize); assertEquals(stats.getAverageStateSize(), avgSize); JsonNode historyNode = rootNode.get("history"); assertNotNull(historyNode); assertTrue(historyNode.isArray()); Iterator<JsonNode> it = historyNode.elements(); for (int i = 0; i < history.size(); i++) { CheckpointStats s = history.get(i); JsonNode node = it.next(); long checkpointId = node.get("id").asLong(); long timestamp = node.get("timestamp").asLong(); long duration = node.get("duration").asLong(); long size = node.get("size").asLong(); assertEquals(s.getCheckpointId(), checkpointId); assertEquals(s.getTriggerTimestamp(), timestamp); assertEquals(s.getDuration(), duration); assertEquals(s.getStateSize(), size); } assertFalse(it.hasNext()); }