ExecutionPlanner { JobGraph createJobGraph(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) { JobGraph jobGraph = new JobGraph(config, appDesc); Set<StreamSpec> sourceStreams = getStreamSpecs(appDesc.getInputStreamIds(), streamConfig); Set<StreamSpec> sinkStreams = getStreamSpecs(appDesc.getOutputStreamIds(), streamConfig); Set<StreamSpec> intermediateStreams = Sets.intersection(sourceStreams, sinkStreams); Set<StreamSpec> inputStreams = Sets.difference(sourceStreams, intermediateStreams); Set<StreamSpec> outputStreams = Sets.difference(sinkStreams, intermediateStreams); Set<TableDescriptor> tables = appDesc.getTableDescriptors(); MapConfig generatedJobConfigs = JobPlanner.generateSingleJobConfig(config); String jobName = generatedJobConfigs.get(JobConfig.JOB_NAME); String jobId = generatedJobConfigs.get(JobConfig.JOB_ID, "1"); JobNode node = jobGraph.getOrCreateJobNode(jobName, jobId); inputStreams.forEach(spec -> jobGraph.addInputStream(spec, node)); outputStreams.forEach(spec -> jobGraph.addOutputStream(spec, node)); intermediateStreams.forEach(spec -> jobGraph.addIntermediateStream(spec, node, node)); for (TableDescriptor table : tables) { jobGraph.addTable(table, node); if (table instanceof LocalTableDescriptor) { LocalTableDescriptor localTable = (LocalTableDescriptor) table; Iterable<String> sideInputs = ListUtils.emptyIfNull(localTable.getSideInputs()); for (String sideInput : sideInputs) { jobGraph.addSideInputStream(getStreamSpec(sideInput, streamConfig)); } } } if (!LegacyTaskApplication.class.isAssignableFrom(appDesc.getAppClass())) { jobGraph.validate(); } return jobGraph; } ExecutionPlanner(Config config, StreamManager streamManager); ExecutionPlan plan(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc);  }
@Test public void testCreateProcessorGraph() { ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamStreamJoin(); JobGraph jobGraph = planner.createJobGraph(graphSpec); assertTrue(jobGraph.getInputStreams().size() == 3); assertTrue(jobGraph.getOutputStreams().size() == 2); assertTrue(jobGraph.getIntermediateStreams().size() == 2); }
@Test public void testCreateJobGraphForTaskApplication() { TaskApplicationDescriptorImpl taskAppDesc = mock(TaskApplicationDescriptorImpl.class); String intermediateStream1 = "intermediate-stream1"; String intermediateBroadcast = "intermediate-broadcast1"; GenericInputDescriptor<KV<Object, Object>> intermediateInput1 = system1Descriptor.getInputDescriptor( intermediateStream1, new KVSerde<>(new NoOpSerde(), new NoOpSerde())); GenericOutputDescriptor<KV<Object, Object>> intermediateOutput1 = system1Descriptor.getOutputDescriptor( intermediateStream1, new KVSerde<>(new NoOpSerde(), new NoOpSerde())); GenericInputDescriptor<KV<Object, Object>> intermediateBroacastInput1 = system1Descriptor.getInputDescriptor( intermediateBroadcast, new KVSerde<>(new NoOpSerde<>(), new NoOpSerde<>())); GenericOutputDescriptor<KV<Object, Object>> intermediateBroacastOutput1 = system1Descriptor.getOutputDescriptor( intermediateBroadcast, new KVSerde<>(new NoOpSerde<>(), new NoOpSerde<>())); inputDescriptors.put(intermediateStream1, intermediateInput1); outputDescriptors.put(intermediateStream1, intermediateOutput1); inputDescriptors.put(intermediateBroadcast, intermediateBroacastInput1); outputDescriptors.put(intermediateBroadcast, intermediateBroacastOutput1); Set<String> broadcastStreams = new HashSet<>(); broadcastStreams.add(intermediateBroadcast); when(taskAppDesc.getInputDescriptors()).thenReturn(inputDescriptors); when(taskAppDesc.getInputStreamIds()).thenReturn(inputDescriptors.keySet()); when(taskAppDesc.getOutputDescriptors()).thenReturn(outputDescriptors); when(taskAppDesc.getOutputStreamIds()).thenReturn(outputDescriptors.keySet()); when(taskAppDesc.getTableDescriptors()).thenReturn(Collections.emptySet()); when(taskAppDesc.getSystemDescriptors()).thenReturn(systemDescriptors); when(taskAppDesc.getIntermediateBroadcastStreamIds()).thenReturn(broadcastStreams); doReturn(MockTaskApplication.class).when(taskAppDesc).getAppClass(); Map<String, String> systemStreamConfigs = new HashMap<>(); inputDescriptors.forEach((key, value) -> systemStreamConfigs.putAll(value.toConfig())); outputDescriptors.forEach((key, value) -> systemStreamConfigs.putAll(value.toConfig())); systemDescriptors.forEach(sd -> systemStreamConfigs.putAll(sd.toConfig())); ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); JobGraph jobGraph = planner.createJobGraph(taskAppDesc); assertEquals(1, jobGraph.getJobNodes().size()); assertTrue(jobGraph.getInputStreams().stream().map(edge -> edge.getName()) .filter(streamId -> inputDescriptors.containsKey(streamId)).collect(Collectors.toList()).isEmpty()); Set<String> intermediateStreams = new HashSet<>(inputDescriptors.keySet()); jobGraph.getInputStreams().forEach(edge -> { if (intermediateStreams.contains(edge.getStreamSpec().getId())) { intermediateStreams.remove(edge.getStreamSpec().getId()); } }); assertEquals(new HashSet<>(Arrays.asList(intermediateStream1, intermediateBroadcast)), intermediateStreams); }
@Test public void testCreateJobGraphForLegacyTaskApplication() { TaskApplicationDescriptorImpl taskAppDesc = mock(TaskApplicationDescriptorImpl.class); when(taskAppDesc.getInputDescriptors()).thenReturn(new HashMap<>()); when(taskAppDesc.getOutputDescriptors()).thenReturn(new HashMap<>()); when(taskAppDesc.getTableDescriptors()).thenReturn(new HashSet<>()); when(taskAppDesc.getSystemDescriptors()).thenReturn(new HashSet<>()); when(taskAppDesc.getIntermediateBroadcastStreamIds()).thenReturn(new HashSet<>()); doReturn(LegacyTaskApplication.class).when(taskAppDesc).getAppClass(); Map<String, String> systemStreamConfigs = new HashMap<>(); inputDescriptors.forEach((key, value) -> systemStreamConfigs.putAll(value.toConfig())); outputDescriptors.forEach((key, value) -> systemStreamConfigs.putAll(value.toConfig())); systemDescriptors.forEach(sd -> systemStreamConfigs.putAll(sd.toConfig())); ExecutionPlanner planner = new ExecutionPlanner(config, streamManager); JobGraph jobGraph = planner.createJobGraph(taskAppDesc); assertEquals(1, jobGraph.getJobNodes().size()); JobNode jobNode = jobGraph.getJobNodes().get(0); assertEquals("test-app", jobNode.getJobName()); assertEquals("test-app-1", jobNode.getJobNameAndId()); assertEquals(0, jobNode.getInEdges().size()); assertEquals(0, jobNode.getOutEdges().size()); assertEquals(0, jobNode.getTables().size()); assertEquals(config, jobNode.getConfig()); }