@Test public void testDisableSlotSharingForIteration() { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Integer> source = env.fromElements(1, 2, 3).name("source"); IterativeStream<Integer> iteration = source.iterate(3000); iteration.name("iteration").setParallelism(2); DataStream<Integer> map = iteration.map(x -> x + 1).name("map").setParallelism(2); DataStream<Integer> filter = map.filter((x) -> false).name("filter").setParallelism(2); iteration.closeWith(filter).print(); List<Transformation<?>> transformations = new ArrayList<>(); transformations.add(source.getTransformation()); transformations.add(iteration.getTransformation()); transformations.add(map.getTransformation()); transformations.add(filter.getTransformation()); StreamGraphGenerator generator = new StreamGraphGenerator(transformations, env.getConfig(), env.getCheckpointConfig()); generator.setSlotSharingEnabled(false); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(generator.generate()); SlotSharingGroup iterationSourceSlotSharingGroup = null; SlotSharingGroup iterationSinkSlotSharingGroup = null; CoLocationGroup iterationSourceCoLocationGroup = null; CoLocationGroup iterationSinkCoLocationGroup = null; for (JobVertex jobVertex : jobGraph.getVertices()) { if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SOURCE_NAME_PREFIX)) { iterationSourceSlotSharingGroup = jobVertex.getSlotSharingGroup(); iterationSourceCoLocationGroup = jobVertex.getCoLocationGroup(); } else if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SINK_NAME_PREFIX)) { iterationSinkSlotSharingGroup = jobVertex.getSlotSharingGroup(); iterationSinkCoLocationGroup = jobVertex.getCoLocationGroup(); } else { assertNull(jobVertex.getSlotSharingGroup()); } } assertNotNull(iterationSourceSlotSharingGroup); assertNotNull(iterationSinkSlotSharingGroup); assertEquals(iterationSourceSlotSharingGroup, iterationSinkSlotSharingGroup); assertNotNull(iterationSourceCoLocationGroup); assertNotNull(iterationSinkCoLocationGroup); assertEquals(iterationSourceCoLocationGroup, iterationSinkCoLocationGroup); }
public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); static JobGraph createJobGraph(StreamGraph streamGraph); static JobGraph createJobGraph(StreamGraph streamGraph, @Nullable JobID jobID); static boolean isChainable(StreamEdge edge, StreamGraph streamGraph); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); static JobGraph createJobGraph(StreamGraph streamGraph); static JobGraph createJobGraph(StreamGraph streamGraph, @Nullable JobID jobID); static boolean isChainable(StreamEdge edge, StreamGraph streamGraph);  }