private <T> WatermarkGaugeExposingOutput<StreamRecord<T>> createOutputCollector(
			StreamTask<?, ?> containingTask,
			StreamConfig operatorConfig,
			Map<Integer, StreamConfig> chainedConfigs,
			ClassLoader userCodeClassloader,
			Map<StreamEdge, RecordWriterOutput<?>> streamOutputs,
			List<StreamOperator<?>> allOperators) {
		List<Tuple2<WatermarkGaugeExposingOutput<StreamRecord<T>>, StreamEdge>> allOutputs = new ArrayList<>(4);

		// create collectors for the network outputs
		for (StreamEdge outputEdge : operatorConfig.getNonChainedOutputs(userCodeClassloader)) {
			@SuppressWarnings("unchecked")
			RecordWriterOutput<T> output = (RecordWriterOutput<T>) streamOutputs.get(outputEdge);

			allOutputs.add(new Tuple2<>(output, outputEdge));
		}

		// Create collectors for the chained outputs
		for (StreamEdge outputEdge : operatorConfig.getChainedOutputs(userCodeClassloader)) {
			int outputId = outputEdge.getTargetId();
			StreamConfig chainedOpConfig = chainedConfigs.get(outputId);

			WatermarkGaugeExposingOutput<StreamRecord<T>> output = createChainedOperator(
				containingTask,
				chainedOpConfig,
				chainedConfigs,
				userCodeClassloader,
				streamOutputs,
				allOperators,
				outputEdge.getOutputTag());
			allOutputs.add(new Tuple2<>(output, outputEdge));
		}

		// if there are multiple outputs, or the outputs are directed, we need to
		// wrap them as one output

		List<OutputSelector<T>> selectors = operatorConfig.getOutputSelectors(userCodeClassloader);

		if (selectors == null || selectors.isEmpty()) {
			// simple path, no selector necessary
			if (allOutputs.size() == 1) {
				return allOutputs.get(0).f0;
			}
			else {
				// send to N outputs. Note that this includes the special case
				// of sending to zero outputs
				@SuppressWarnings({"unchecked", "rawtypes"})
				Output<StreamRecord<T>>[] asArray = new Output[allOutputs.size()];
				for (int i = 0; i < allOutputs.size(); i++) {
					asArray[i] = allOutputs.get(i).f0;
				}

				// This is the inverse of creating the normal ChainingOutput.
				// If the chaining output does not copy we need to copy in the broadcast output,
				// otherwise multi-chaining would not work correctly.
				if (containingTask.getExecutionConfig().isObjectReuseEnabled()) {
					return new CopyingBroadcastingOutputCollector<>(asArray, this);
				} else  {
					return new BroadcastingOutputCollector<>(asArray, this);
				}
			}
		}
		else {
			// selector present, more complex routing necessary

			// This is the inverse of creating the normal ChainingOutput.
			// If the chaining output does not copy we need to copy in the broadcast output,
			// otherwise multi-chaining would not work correctly.
			if (containingTask.getExecutionConfig().isObjectReuseEnabled()) {
				return new CopyingDirectedOutput<>(selectors, allOutputs);
			} else {
				return new DirectedOutput<>(selectors, allOutputs);
			}

		}
	}
private < T > WatermarkGaugeExposingOutput < StreamRecord < T > > createOutputCollector ( StreamTask < ? , ? > containingTask , StreamConfig operatorConfig , Map < Integer , StreamConfig > chainedConfigs , ClassLoader userCodeClassloader , Map < StreamEdge , RecordWriterOutput < ? > > streamOutputs , List < StreamOperator < ? > > allOperators ) { List < Tuple2 < WatermarkGaugeExposingOutput < StreamRecord < T > > , StreamEdge > > allOutputs = new ArrayList < > ( 4 ) ; for ( StreamEdge outputEdge : operatorConfig . getNonChainedOutputs ( userCodeClassloader ) ) { @ SuppressWarnings ( "unchecked" ) RecordWriterOutput < T > output = ( RecordWriterOutput < T > ) streamOutputs . get ( outputEdge ) ; allOutputs . add ( new Tuple2 < > ( output , outputEdge ) ) ; } for ( StreamEdge outputEdge : operatorConfig . getChainedOutputs ( userCodeClassloader ) ) { int outputId = outputEdge . getTargetId ( ) ; StreamConfig chainedOpConfig = chainedConfigs . get ( outputId ) ; WatermarkGaugeExposingOutput < StreamRecord < T > > output = createChainedOperator ( containingTask , chainedOpConfig , chainedConfigs , userCodeClassloader , streamOutputs , allOperators , outputEdge . getOutputTag ( ) ) ; allOutputs . add ( new Tuple2 < > ( output , outputEdge ) ) ; } List < OutputSelector < T > > selectors = operatorConfig . getOutputSelectors ( userCodeClassloader ) ; if ( selectors == null || selectors . isEmpty ( ) ) { if ( allOutputs . size ( ) == 1 ) { return allOutputs . get ( 0 ) . f0 ; } else { @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) Output < StreamRecord < T > > [ ] asArray = new Output [ allOutputs . size ( ) ] ; for ( int i = 0 ; i < allOutputs . size ( ) ; i ++ ) { asArray [ i ] = allOutputs . get ( i ) . f0 ; } if ( containingTask . getExecutionConfig ( ) . isObjectReuseEnabled ( ) ) { return new CopyingBroadcastingOutputCollector < > ( asArray , this ) ; } else { return new BroadcastingOutputCollector < > ( asArray , this ) ; } } } else { if ( containingTask . getExecutionConfig ( ) . isObjectReuseEnabled ( ) ) { return new CopyingDirectedOutput < > ( selectors , allOutputs ) ; } else { return new DirectedOutput < > ( selectors , allOutputs ) ; } } }
------------------------------------------------------------------------