JobGraphJsonGenerator { String toJson(JobGraph jobGraph) throws Exception { JobGraphJson jobGraphJson = new JobGraphJson(); ApplicationConfig appConfig = jobGraph.getApplicationConfig(); jobGraphJson.applicationName = appConfig.getAppName(); jobGraphJson.applicationId = appConfig.getAppId(); jobGraphJson.sourceStreams = new HashMap<>(); jobGraphJson.sinkStreams = new HashMap<>(); jobGraphJson.intermediateStreams = new HashMap<>(); jobGraphJson.tables = new HashMap<>(); jobGraph.getInputStreams().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sourceStreams)); jobGraph.getOutputStreams().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sinkStreams)); jobGraph.getIntermediateStreamEdges().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.intermediateStreams)); jobGraph.getTables().forEach(t -> buildTableJson(t, jobGraphJson.tables)); jobGraphJson.jobs = jobGraph.getJobNodes().stream() .map(this::buildJobNodeJson) .collect(Collectors.toList()); ByteArrayOutputStream out = new ByteArrayOutputStream(); ObjectMapper mapper = new ObjectMapper(); mapper.writeValue(out, jobGraphJson); return new String(out.toByteArray()); }    }
@Test public void testTaskApplication() throws Exception { JobGraphJsonGenerator jsonGenerator = new JobGraphJsonGenerator(); JobGraph mockJobGraph = mock(JobGraph.class); ApplicationConfig mockAppConfig = mock(ApplicationConfig.class); when(mockAppConfig.getAppName()).thenReturn("testTaskApp"); when(mockAppConfig.getAppId()).thenReturn("testTaskAppId"); when(mockJobGraph.getApplicationConfig()).thenReturn(mockAppConfig); Set<StreamEdge> inEdges = new HashSet<>(mockJobNode.getInEdges().values()); Set<StreamEdge> outEdges = new HashSet<>(mockJobNode.getOutEdges().values()); Set<StreamEdge> intermediateEdges = new HashSet<>(inEdges); intermediateEdges.retainAll(outEdges); inEdges.removeAll(intermediateEdges); outEdges.removeAll(intermediateEdges); when(mockJobGraph.getInputStreams()).thenReturn(inEdges); when(mockJobGraph.getOutputStreams()).thenReturn(outEdges); when(mockJobGraph.getIntermediateStreamEdges()).thenReturn(intermediateEdges); Set<TableDescriptor> tables = new HashSet<>(mockJobNode.getTables().values()); when(mockJobGraph.getTables()).thenReturn(tables); when(mockJobGraph.getJobNodes()).thenReturn(Collections.singletonList(mockJobNode)); String graphJson = jsonGenerator.toJson(mockJobGraph); ObjectMapper objectMapper = new ObjectMapper(); JobGraphJsonGenerator.JobGraphJson jsonObject = objectMapper.readValue(graphJson.getBytes(), JobGraphJsonGenerator.JobGraphJson.class); assertEquals("testTaskAppId", jsonObject.applicationId); assertEquals("testTaskApp", jsonObject.applicationName); Set<String> inStreamIds = inEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet()); assertThat(jsonObject.sourceStreams.keySet(), Matchers.containsInAnyOrder(inStreamIds.toArray())); Set<String> outStreamIds = outEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet()); assertThat(jsonObject.sinkStreams.keySet(), Matchers.containsInAnyOrder(outStreamIds.toArray())); Set<String> intStreamIds = intermediateEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet()); assertThat(jsonObject.intermediateStreams.keySet(), Matchers.containsInAnyOrder(intStreamIds.toArray())); Set<String> tableIds = tables.stream().map(t -> t.getTableId()).collect(Collectors.toSet()); assertThat(jsonObject.tables.keySet(), Matchers.containsInAnyOrder(tableIds.toArray())); JobGraphJsonGenerator.JobNodeJson expectedNodeJson = new JobGraphJsonGenerator.JobNodeJson(); expectedNodeJson.jobId = mockJobNode.getJobId(); expectedNodeJson.jobName = mockJobNode.getJobName(); assertEquals(1, jsonObject.jobs.size()); JobGraphJsonGenerator.JobNodeJson actualNodeJson = jsonObject.jobs.get(0); assertEquals(expectedNodeJson.jobId, actualNodeJson.jobId); assertEquals(expectedNodeJson.jobName, actualNodeJson.jobName); assertEquals(3, actualNodeJson.operatorGraph.inputStreams.size()); assertEquals(2, actualNodeJson.operatorGraph.outputStreams.size()); assertEquals(0, actualNodeJson.operatorGraph.operators.size()); }
@Test public void testLegacyTaskApplication() throws Exception { JobGraphJsonGenerator jsonGenerator = new JobGraphJsonGenerator(); JobGraph mockJobGraph = mock(JobGraph.class); ApplicationConfig mockAppConfig = mock(ApplicationConfig.class); when(mockAppConfig.getAppName()).thenReturn("testTaskApp"); when(mockAppConfig.getAppId()).thenReturn("testTaskAppId"); when(mockJobGraph.getApplicationConfig()).thenReturn(mockAppConfig); String graphJson = jsonGenerator.toJson(mockJobGraph); ObjectMapper objectMapper = new ObjectMapper(); JobGraphJsonGenerator.JobGraphJson jsonObject = objectMapper.readValue(graphJson.getBytes(), JobGraphJsonGenerator.JobGraphJson.class); assertEquals("testTaskAppId", jsonObject.applicationId); assertEquals("testTaskApp", jsonObject.applicationName); JobGraphJsonGenerator.JobNodeJson expectedNodeJson = new JobGraphJsonGenerator.JobNodeJson(); expectedNodeJson.jobId = mockJobNode.getJobId(); expectedNodeJson.jobName = mockJobNode.getJobName(); assertEquals(0, jsonObject.jobs.size()); }