flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
/**
	 * Transforms a {@code FeedbackTransformation}.
	 *
	 * <p>This will recursively transform the input and the feedback edges. We return the
	 * concatenation of the input IDs and the feedback IDs so that downstream operations can be
	 * wired to both.
	 *
	 * <p>This is responsible for creating the IterationSource and IterationSink which are used to
	 * feed back the elements.
	 */


This is responsible for creating the IterationSource and IterationSink which are used to
feed back the elements.
This is responsible for creating the IterationSource and IterationSink which are used to feed back the elements .
private <T> Collection<Integer> transformFeedback(FeedbackTransformation<T> iterate) {
		if (iterate.getFeedbackEdges().size() <= 0) {
			throw new IllegalStateException("Iteration " + iterate + " does not have any feedback edges.");
		}
		StreamTransformation<T> input = iterate.getInput();
		List<Integer> resultIds = new ArrayList<>();
		Collection<Integer> inputIds = transform(input);
		resultIds.addAll(inputIds);
		if (alreadyTransformed.containsKey(iterate)) {
			return alreadyTransformed.get(iterate);
		}
		Tuple2<StreamNode, StreamNode> itSourceAndSink = streamGraph.createIterationSourceAndSink(
			iterate.getId(),
			getNewIterationNodeId(),
			getNewIterationNodeId(),
			iterate.getWaitTime(),
			iterate.getParallelism(),
			iterate.getMaxParallelism(),
			iterate.getMinResources(),
			iterate.getPreferredResources());
		StreamNode itSource = itSourceAndSink.f0;
		StreamNode itSink = itSourceAndSink.f1;
		streamGraph.setSerializers(itSource.getId(), null, null, iterate.getOutputType().createSerializer(env.getConfig()));
		streamGraph.setSerializers(itSink.getId(), iterate.getOutputType().createSerializer(env.getConfig()), null, null);
		resultIds.add(itSource.getId());
		alreadyTransformed.put(iterate, resultIds);
		List<Integer> allFeedbackIds = new ArrayList<>();
		for (StreamTransformation<T> feedbackEdge : iterate.getFeedbackEdges()) {
			Collection<Integer> feedbackIds = transform(feedbackEdge);
			allFeedbackIds.addAll(feedbackIds);
			for (Integer feedbackId: feedbackIds) {
				streamGraph.addEdge(feedbackId,
						itSink.getId(),
						0
				);
			}
		}
		String slotSharingGroup = determineSlotSharingGroup(null, allFeedbackIds);
		itSink.setSlotSharingGroup(slotSharingGroup);
		itSource.setSlotSharingGroup(slotSharingGroup);
		return resultIds;
	}
private < T > Collection < Integer > transformFeedback ( FeedbackTransformation < T > iterate ) { if ( iterate . getFeedbackEdges ( ) . size ( ) <= 0 ) { throw new IllegalStateException ( "Iteration " + iterate + " does not have any feedback edges." ) ; } StreamTransformation < T > input = iterate . getInput ( ) ; List < Integer > resultIds = new ArrayList < > ( ) ; Collection < Integer > inputIds = transform ( input ) ; resultIds . addAll ( inputIds ) ; if ( alreadyTransformed . containsKey ( iterate ) ) { return alreadyTransformed . get ( iterate ) ; } Tuple2 < StreamNode , StreamNode > itSourceAndSink = streamGraph . createIterationSourceAndSink ( iterate . getId ( ) , getNewIterationNodeId ( ) , getNewIterationNodeId ( ) , iterate . getWaitTime ( ) , iterate . getParallelism ( ) , iterate . getMaxParallelism ( ) , iterate . getMinResources ( ) , iterate . getPreferredResources ( ) ) ; StreamNode itSource = itSourceAndSink . f0 ; StreamNode itSink = itSourceAndSink . f1 ; streamGraph . setSerializers ( itSource . getId ( ) , null , null , iterate . getOutputType ( ) . createSerializer ( env . getConfig ( ) ) ) ; streamGraph . setSerializers ( itSink . getId ( ) , iterate . getOutputType ( ) . createSerializer ( env . getConfig ( ) ) , null , null ) ; resultIds . add ( itSource . getId ( ) ) ; alreadyTransformed . put ( iterate , resultIds ) ; List < Integer > allFeedbackIds = new ArrayList < > ( ) ; for ( StreamTransformation < T > feedbackEdge : iterate . getFeedbackEdges ( ) ) { Collection < Integer > feedbackIds = transform ( feedbackEdge ) ; allFeedbackIds . addAll ( feedbackIds ) ; for ( Integer feedbackId : feedbackIds ) { streamGraph . addEdge ( feedbackId , itSink . getId ( ) , 0 ) ; } } String slotSharingGroup = determineSlotSharingGroup ( null , allFeedbackIds ) ; itSink . setSlotSharingGroup ( slotSharingGroup ) ; itSource . setSlotSharingGroup ( slotSharingGroup ) ; return resultIds ; }
Transforms a {@code FeedbackTransformation}.
Transforms a { @code FeedbackTransformation } .
// first transform the input stream(s) and store the result IDs // the recursive transform might have already transformed this // create the fake iteration source/sink pair // We set the proper serializers for the sink/source // also add the feedback source ID to the result IDs, so that downstream operators will // add both as input // at the iterate to the already-seen-set with the result IDs, so that we can transform // the feedback edges and let them stop when encountering the iterate node // so that we can determine the slot sharing group from all feedback edges