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 CoFeedbackTransformation}.
	 *
	 * <p>This will only transform feedback edges, the result of this transform will be wired
	 * to the second input of a Co-Transform. The original input is wired directly to the first
	 * input of the downstream Co-Transform.
	 *
	 * <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 <F> Collection<Integer> transformCoFeedback(CoFeedbackTransformation<F> coIterate) {
		Tuple2<StreamNode, StreamNode> itSourceAndSink = streamGraph.createIterationSourceAndSink(
				coIterate.getId(),
				getNewIterationNodeId(),
				getNewIterationNodeId(),
				coIterate.getWaitTime(),
				coIterate.getParallelism(),
				coIterate.getMaxParallelism(),
				coIterate.getMinResources(),
				coIterate.getPreferredResources());
		StreamNode itSource = itSourceAndSink.f0;
		StreamNode itSink = itSourceAndSink.f1;
		streamGraph.setSerializers(itSource.getId(), null, null, coIterate.getOutputType().createSerializer(env.getConfig()));
		streamGraph.setSerializers(itSink.getId(), coIterate.getOutputType().createSerializer(env.getConfig()), null, null);
		Collection<Integer> resultIds = Collections.singleton(itSource.getId());
		alreadyTransformed.put(coIterate, resultIds);
		List<Integer> allFeedbackIds = new ArrayList<>();
		for (StreamTransformation<F> feedbackEdge : coIterate.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 Collections.singleton(itSource.getId());
	}
private < F > Collection < Integer > transformCoFeedback ( CoFeedbackTransformation < F > coIterate ) { Tuple2 < StreamNode , StreamNode > itSourceAndSink = streamGraph . createIterationSourceAndSink ( coIterate . getId ( ) , getNewIterationNodeId ( ) , getNewIterationNodeId ( ) , coIterate . getWaitTime ( ) , coIterate . getParallelism ( ) , coIterate . getMaxParallelism ( ) , coIterate . getMinResources ( ) , coIterate . getPreferredResources ( ) ) ; StreamNode itSource = itSourceAndSink . f0 ; StreamNode itSink = itSourceAndSink . f1 ; streamGraph . setSerializers ( itSource . getId ( ) , null , null , coIterate . getOutputType ( ) . createSerializer ( env . getConfig ( ) ) ) ; streamGraph . setSerializers ( itSink . getId ( ) , coIterate . getOutputType ( ) . createSerializer ( env . getConfig ( ) ) , null , null ) ; Collection < Integer > resultIds = Collections . singleton ( itSource . getId ( ) ) ; alreadyTransformed . put ( coIterate , resultIds ) ; List < Integer > allFeedbackIds = new ArrayList < > ( ) ; for ( StreamTransformation < F > feedbackEdge : coIterate . 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 Collections . singleton ( itSource . getId ( ) ) ; }
Transforms a {@code CoFeedbackTransformation}.
Transforms a { @code CoFeedbackTransformation } .
// For Co-Iteration we don't need to transform the input and wire the input to the // head operator by returning the input IDs, the input is directly wired to the left // input of the co-operation. This transform only needs to return the ids of the feedback // edges, since they need to be wired to the second input of the co-operation. // create the fake iteration source/sink pair // We set the proper serializers for the sink/source // 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