flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
/**
	 * Adds a new virtual node that is used to connect a downstream vertex to only the outputs with
	 * the selected side-output {@link OutputTag}.
	 *
	 * @param originalId ID of the node that should be connected to.
	 * @param virtualId ID of the virtual node.
	 * @param outputTag The selected side-output {@code OutputTag}.
	 */
Adds a new virtual node that is used to connect a downstream vertex to only the outputs with
the selected side-output  OutputTag.

@param originalId ID of the node that should be connected to.
@param virtualId ID of the virtual node.
@param outputTag The selected side-output  OutputTag.
Adds a new virtual node that is used to connect a downstream vertex to only the outputs with the selected side - output OutputTag . @param originalId ID of the node that should be connected to . @param virtualId ID of the virtual node . @param outputTag The selected side - output OutputTag .
public void addVirtualSideOutputNode(Integer originalId, Integer virtualId, OutputTag outputTag) {
		if (virtualSideOutputNodes.containsKey(virtualId)) {
			throw new IllegalStateException("Already has virtual output node with id " + virtualId);
		}
		for (Tuple2<Integer, OutputTag> tag : virtualSideOutputNodes.values()) {
			if (!tag.f0.equals(originalId)) {
				continue;
			}
			if (tag.f1.getId().equals(outputTag.getId()) &&
					!tag.f1.getTypeInfo().equals(outputTag.getTypeInfo())) {
				throw new IllegalArgumentException("Trying to add a side output for the same " +
						"side-output id with a different type. This is not allowed. Side-output ID: " +
						tag.f1.getId());
			}
		}
		virtualSideOutputNodes.put(virtualId, new Tuple2<>(originalId, outputTag));
	}
public void addVirtualSideOutputNode ( Integer originalId , Integer virtualId , OutputTag outputTag ) { if ( virtualSideOutputNodes . containsKey ( virtualId ) ) { throw new IllegalStateException ( "Already has virtual output node with id " + virtualId ) ; } for ( Tuple2 < Integer , OutputTag > tag : virtualSideOutputNodes . values ( ) ) { if ( ! tag . f0 . equals ( originalId ) ) { continue ; } if ( tag . f1 . getId ( ) . equals ( outputTag . getId ( ) ) && ! tag . f1 . getTypeInfo ( ) . equals ( outputTag . getTypeInfo ( ) ) ) { throw new IllegalArgumentException ( "Trying to add a side output for the same " + "side-output id with a different type. This is not allowed. Side-output ID: " + tag . f1 . getId ( ) ) ; } } virtualSideOutputNodes . put ( virtualId , new Tuple2 < > ( originalId , outputTag ) ) ; }
Adds a new virtual node that is used to connect a downstream vertex to only the outputs with
 the selected side-output {@link OutputTag}.
Adds a new virtual node that is used to connect a downstream vertex to only the outputs with the selected side - output { @link OutputTag } .
// verify that we don't already have a virtual node for the given originalId/outputTag // combination with a different TypeInformation. This would indicate that someone is trying // to read a side output from an operation with a different type for the same side output // id. // different source operator