docs: 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);
  			}
  
  		}
  	}